FEATURE: added files for mysql

alphamind/data/dbmodel/models_mysql.py

+"""
+Created on 2020-10-11
+
+@author: cheng.li
+"""
+
+from sqlalchemy import (
+    Column,
+    DECIMAL,
+    Date,
+    DateTime,
+    Float,
+    Index,
+    String,
+    TIMESTAMP,
+    Table,
+    Text,
+    VARBINARY,
+    text
+)
+from sqlalchemy.dialects.mysql import (
+    BIGINT,
+    INTEGER,
+    SET,
+    TIMESTAMP
+)
+from sqlalchemy.ext.declarative import declarative_base
+
+Base = declarative_base()
+metadata = Base.metadata
+
+
+class _StkDailyPricePro(Base):
+    __tablename__ = 'stk_daily_price_pro'
+    __table_args__ = (
+        Index('unique_stk_daily_price_pro_index', 'trade_date', 'security_code', 'flag', unique=True),
+    )
+
+    id = Column(INTEGER(10), primary_key=True)
+    trade_date = Column(Date)
+    code = Column("security_code", String(20))
+    exchange = Column(String(10))
+    security_type = Column(String(10))
+    symbol = Column(String(20))
+    name = Column(String(100))
+    cur = Column(String(10))
+    pre_close = Column(DECIMAL(20, 4))
+    open = Column(DECIMAL(20, 4))
+    high = Column(DECIMAL(20, 4))
+    low = Column(DECIMAL(20, 4))
+    close = Column(DECIMAL(20, 4))
+    volume = Column(DECIMAL(20, 0))
+    money = Column(DECIMAL(20, 3))
+    deals = Column(DECIMAL(20, 0))
+    avg_price = Column(DECIMAL(20, 4))
+    avg_vol = Column(DECIMAL(20, 4))
+    change = Column(DECIMAL(20, 4))
+    chgPct = Column("change_pct", DECIMAL(20, 4))
+    amplitude = Column(DECIMAL(20, 4))
+    amplitude_lower = Column(DECIMAL(20, 4))
+    fx_rate = Column(DECIMAL(16, 6))
+    capitalization = Column(DECIMAL(19, 4))
+    tot_market_cap_cur = Column(DECIMAL(30, 8))
+    tot_market_cap = Column(DECIMAL(30, 8))
+    circulating_cap = Column(DECIMAL(19, 4))
+    circulating_market_cap_cur = Column(DECIMAL(30, 8))
+    circulating_market_cap = Column(DECIMAL(30, 8))
+    restrict_circulating_cap = Column(DECIMAL(19, 4))
+    restrict_circulating_market_cap_cur = Column(DECIMAL(30, 8))
+    restrict_circulating_market_cap = Column(DECIMAL(30, 8))
+    non_circulating_cap = Column(DECIMAL(19, 4))
+    calc_non_circulating_cap = Column(DECIMAL(19, 4))
+    no_calc_non_circulating_cap = Column(DECIMAL(19, 4))
+    calc_non_circulating_market_cap_cur = Column(DECIMAL(30, 8))
+    turn_rate = Column(DECIMAL(10, 4))
+    trade_status = Column(String(10))
+    is_count = Column(INTEGER(11))
+    upper_limit_price = Column(DECIMAL(20, 4))
+    lower_limit_price = Column(DECIMAL(20, 4))
+    change_pct_status = Column(String(1), nullable=False)
+    special_trade_type = Column(String(2))
+    is_valid = Column(INTEGER(11), nullable=False)
+    entry_date = Column(DateTime, nullable=False)
+    entry_time = Column(String(8), nullable=False)
+    lat_factor = Column(DECIMAL(29, 16))
+    flag = Column(INTEGER(11), index=True, server_default=text("'1'"))
+    is_verify = Column(INTEGER(11), index=True, server_default=text("'0'"))
+    create_time = Column(TIMESTAMP, server_default=text("CURRENT_TIMESTAMP"))
+    update_time = Column(TIMESTAMP, index=True, server_default=text("CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP"))
+
+
+
+Market = _StkDailyPricePro

alphamind/data/engines/sqlengine_mysql.py

+# -*- coding: utf-8 -*-
+"""
+Created on 2020-10-11
+
+@author: cheng.li
+"""
+
+from typing import Iterable
+
+
+import pandas as pd
+import sqlalchemy as sa
+import sqlalchemy.orm as orm
+from sqlalchemy import (
+    and_,
+    select
+)
+from sqlalchemy.sql import func
+
+from PyFin.api import advanceDateByCalendar
+
+from alphamind.data.dbmodel.models_mysql import (
+    Market
+)
+from alphamind.data.processing import factor_processing
+
+
+DAILY_RETURN_OFFSET = 0
+
+
+class SqlEngine:
+
+    def __init__(self, db_url: str):
+        self._engine = sa.create_engine(db_url)
+        self._session = self.create_session()
+
+    def __del__(self):
+        if self._session:
+            self._session.close()
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        if self._session:
+            self._session.close()
+
+    def create_session(self):
+        db_session = orm.sessionmaker(bind=self._engine)
+        return db_session()
+
+    # def _create_stats(self, table, horizon, offset, code_attr='security_code'):
+    #     stats = func.sum(self._ln_func(1. + table.change_pct)).over(
+    #         partition_by=getattr(table, code_attr),
+    #         order_by=table.trade_date,
+    #         rows=(
+    #         1 + DAILY_RETURN_OFFSET + offset, 1 + horizon + DAILY_RETURN_OFFSET + offset)).label(
+    #         'dx')
+    #     return stats
+
+    def fetch_dx_return(self,
+                        ref_date: str,
+                        codes: Iterable[int],
+                        expiry_date: str = None,
+                        horizon: int = 0,
+                        offset: int = 0,
+                        neutralized_risks: list = None,
+                        pre_process=None,
+                        post_process=None) -> pd.DataFrame:
+        start_date = ref_date
+
+        if not expiry_date:
+            end_date = advanceDateByCalendar('china.sse', ref_date,
+                                             str(
+                                                 1 + horizon + offset + DAILY_RETURN_OFFSET) + 'b').strftime(
+                '%Y-%m-%d')
+        else:
+            end_date = expiry_date
+
+        query = select([Market.trade_date, Market.code, Market.chgPct]).where(
+            and_(
+                Market.trade_date.between(start_date, end_date),
+                Market.code.in_(codes)
+            )
+        )
+
+        df = pd.read_sql(query, self._session.bind).dropna()
+        df = df[df.trade_date == ref_date]
+
+        if neutralized_risks:
+            _, risk_exp = self.fetch_risk_model(ref_date, codes)
+            df = pd.merge(df, risk_exp, on='code').dropna()
+            df[['dx']] = factor_processing(df[['dx']].values,
+                                           pre_process=pre_process,
+                                           risk_factors=df[neutralized_risks].values,
+                                           post_process=post_process)
+
+        return df # df[['code', 'dx']]
+
+
+if __name__ == "__main__":
+    db_url = "mysql+mysqldb://reader:Reader#2020@121.37.138.1:13317/vision?charset=utf8"
+
+    sql_engine = SqlEngine(db_url=db_url)
+    df = sql_engine.fetch_dx_return(ref_date='2020-09-29', codes=["2010003704"])
+    print(df)