评价修改

app/api/engine.py

@@ -57,6 +57,18 @@ tamp_user_engine = create_engine(
     ),
     echo=True
 )
+
+tamp_fund_engine = create_engine(
+    'mysql+pymysql://{user}:{password}@{host}:{port}/{db}?charset={charset}'.format(
+        db=config[env]['MySQL']['tamp_fund_db'],
+        host=config[env]['MySQL']['host'],
+        port=config[env]['MySQL']['port'],
+        user=config[env]['MySQL']['user'],
+        password=config[env]['MySQL']['password'],
+        charset="utf8"
+    ),
+    echo=True
+)
 # tamp_product_session = scoped_session(sessionmaker(bind=tamp_product_engine))()
 # tamp_order_session = scoped_session(sessionmaker(bind=tamp_order_engine))()
 # tamp_user_session = scoped_session(sessionmaker(bind=tamp_user_engine))()

app/config/config.yaml

@@ -43,6 +43,7 @@ prod:
     tamp_product_db: tamp_product
     tamp_order_db: tamp_order
     tamp_user_db: tamp_user
+    tamp_fund_db: tamp_fund
     host: tamper.mysql.polardb.rds.aliyuncs.com
     port: 3306
     user: tamp_admin

app/utils/fund_rank.py

-# import pymysql
+
 from sqlalchemy import create_engine
 
-db = create_engine(
-    'mysql+pymysql://tamp_fund:@imeng408@tamper.mysql.polardb.rds.aliyuncs.com:3306/tamp_fund?charset=utf8mb4',
-    pool_size=50,
-    pool_recycle=3600,
-    pool_pre_ping=True)
-con = db.connect()
 
-import logging
+# db = create_engine(
+#     'mysql+pymysql://tamp_fund:@imeng408@tamper.mysql.polardb.rds.aliyuncs.com:3306/tamp_fund?charset=utf8mb4',
+#     pool_size=50,
+#     pool_recycle=3600,
+#     pool_pre_ping=True)
+# con = db.connect()
 
+import logging
 logging.basicConfig(level=logging.INFO)
 
+from app.api.engine import tamp_fund_engine, TAMP_SQL
 from app.utils.week_evaluation import *
 
 
@@ -35,32 +36,39 @@ def get_nav(fund, start_date, rollback=False, invest_type='public'):
     Returns:df[DataFrame]: 索引为净值公布日， 列为复权净值的净值表; 查询失败则返回None
 
     """
-    if invest_type == 'public':
-        sql = "SELECT ts_code, end_date, adj_nav FROM public_fund_nav " \
-              "WHERE ts_code='{}'".format(fund)
-        df = pd.read_sql(sql, con).dropna(how='any')
-        df.rename({'ts_code': 'fund_id'}, axis=1, inplace=True)
-    else:
-        sql = "SELECT fund_id, price_date, cumulative_nav FROM fund_nav " \
-              "WHERE fund_id='{}'".format(fund)
-        df = pd.read_sql(sql, con).dropna(how='any')
-        df.rename({'price_date': 'end_date', 'cumulative_nav': 'adj_nav'}, axis=1, inplace=True)
-
-    if df['adj_nav'].count() == 0:
-        logging.log(logging.ERROR, "CAN NOT FIND {}".format(fund))
-        return None
-
-    df['end_date'] = pd.to_datetime(df['end_date'])
-
-    if rollback and df['end_date'].min() < start_date < df['end_date'].max():
-        while start_date not in list(df['end_date']):
-            start_date -= datetime.timedelta(days=1)
-
-    df = df[df['end_date'] >= start_date]
-    df.drop_duplicates(subset='end_date', inplace=True, keep='first')
-    df.set_index('end_date', inplace=True)
-    df.sort_index(inplace=True, ascending=True)
-    return df
+    with TAMP_SQL(tamp_fund_engine) as tamp_product:
+        tamp_product_session = tamp_product.session
+        if invest_type == 'public':
+            sql = "SELECT ts_code, end_date, adj_nav FROM public_fund_nav " \
+                  "WHERE ts_code='{}'".format(fund)
+            cur = tamp_product_session.execute(sql)
+            data = cur.fetchall()
+            df = pd.DataFrame(list(data), columns=['ts_code', 'end_date', 'adj_nav']).dropna(how='any')
+            df.rename({'ts_code': 'fund_id'}, axis=1, inplace=True)
+        else:
+            sql = "SELECT fund_id, price_date, cumulative_nav FROM fund_nav " \
+                  "WHERE fund_id='{}'".format(fund)
+            # df = pd.read_sql(sql, con).dropna(how='any')
+            cur = tamp_product_session.execute(sql)
+            data = cur.fetchall()
+            df = pd.DataFrame(data, columns=['fund_id', 'price_date', 'cumulative_nav']).dropna(how='any')
+            df.rename({'price_date': 'end_date', 'cumulative_nav': 'adj_nav'}, axis=1, inplace=True)
+
+        if df['adj_nav'].count() == 0:
+            logging.log(logging.ERROR, "CAN NOT FIND {}".format(fund))
+            return None
+
+        df['end_date'] = pd.to_datetime(df['end_date'])
+
+        if rollback and df['end_date'].min() < start_date < df['end_date'].max():
+            while start_date not in list(df['end_date']):
+                start_date -= datetime.timedelta(days=1)
+
+        df = df[df['end_date'] >= start_date]
+        df.drop_duplicates(subset='end_date', inplace=True, keep='first')
+        df.set_index('end_date', inplace=True)
+        df.sort_index(inplace=True, ascending=True)
+        return df
 
 
 def get_frequency(df):
@@ -97,11 +105,16 @@ def get_trade_cal():
     Returns:df[DataFrame]: 索引为交易日， 列为交易日的上交所交易日历表
 
     """
-    sql = 'SELECT cal_date FROM stock_trade_cal WHERE is_open=1'
-    df = pd.read_sql(sql, con)
-    df['end_date'] = pd.to_datetime(df['cal_date'])
-    df.set_index('end_date', drop=False, inplace=True)
-    return df
+    with TAMP_SQL(tamp_fund_engine) as tamp_product:
+        tamp_product_session = tamp_product.session
+        sql = 'SELECT cal_date FROM stock_trade_cal WHERE is_open=1'
+        cur = tamp_product_session.execute(sql)
+        data = cur.fetchall()
+        df = pd.DataFrame(list(data), columns=['cal_date']).dropna(how='all')
+        # df = pd.read_sql(sql, con)
+        df['end_date'] = pd.to_datetime(df['cal_date'])
+        df.set_index('end_date', drop=False, inplace=True)
+        return df
 
 
 def get_manager(invest_type):
@@ -113,13 +126,21 @@ def get_manager(invest_type):
     Returns:
 
     """
-    if invest_type == 'public':
-        sql = 'SELECT ts_code, name FROM public_fund_manager WHERE end_date IS NULL'
-        df = pd.read_sql(sql, con)
-    else:
-        sql = 'SELECT fund_id, fund_manager_id FROM fund_manager_mapping'
-        df = pd.read_sql(sql, con)
-    return df
+    with TAMP_SQL(tamp_fund_engine) as tamp_product:
+        tamp_product_session = tamp_product.session
+        if invest_type == 'public':
+            sql = 'SELECT ts_code, name FROM public_fund_manager WHERE end_date IS NULL'
+            # df = pd.read_sql(sql, con)
+            cur = tamp_product_session.execute(sql)
+            data = cur.fetchall()
+            df = pd.DataFrame(list(data), columns=['ts_code', 'name'])
+        else:
+            sql = 'SELECT fund_id, fund_manager_id FROM fund_manager_mapping'
+            # df = pd.read_sql(sql, con)
+            cur = tamp_product_session.execute(sql)
+            data = cur.fetchall()
+            df = pd.DataFrame(list(data), columns=['fund_id', 'fund_manager_id'])
+        return df
 
 
 def get_fund_info(end_date, invest_type):
@@ -132,23 +153,33 @@ def get_fund_info(end_date, invest_type):
     Returns:
         [type]: [description]
     """
-    if invest_type == 'public':
-        sql = "SELECT ts_code, fund_type, management FROM public_fund_basic " \
-              "WHERE delist_date IS NULL AND (due_date IS NULL OR due_date>'{}')".format(end_date.strftime('%Y%m%d'))
-        df = pd.read_sql(sql, con).dropna(how='all')
-        manager_info = get_manager(invest_type)
-
-        df.rename({'ts_code': 'fund_id'}, axis=1, inplace=True)
-        df = pd.merge(df, manager_info, how="left", on='fund_id')
-    else:
-        sql = "SELECT id, substrategy FROM fund_info WHERE delete_tag=0 " \
-              "AND substrategy!=-1"
-        df = pd.read_sql(sql, con).dropna(how='all')
+    with TAMP_SQL(tamp_fund_engine) as tamp_product:
+        tamp_product_session = tamp_product.session
+        if invest_type == 'public':
+            sql = "SELECT ts_code, fund_type, management FROM public_fund_basic " \
+                  "WHERE delist_date IS NULL AND (due_date IS NULL OR due_date>'{}')".format(end_date.strftime('%Y%m%d'))
+            # df = pd.read_sql(sql, con).dropna(how='all')
+            cur = tamp_product_session.execute(sql)
+            data = cur.fetchall()
+
+            df = pd.DataFrame(list(data), columns=['ts_code', 'fund_type', 'management'])
+            manager_info = get_manager(invest_type)
+
+            df.rename({'ts_code': 'fund_id'}, axis=1, inplace=True)
+            df = pd.merge(df, manager_info, how="left", on='fund_id')
+        else:
 
-        df.rename({'id': 'fund_id'}, axis=1, inplace=True)
-        manager_info = get_manager(invest_type)
-        df = pd.merge(df, manager_info, how="inner", on='fund_id')
-    return df
+            sql = "SELECT id, substrategy FROM fund_info WHERE delete_tag=0 " \
+                  "AND substrategy!=-1"
+            cur = tamp_product_session.execute(sql)
+            data = cur.fetchall()
+            df = pd.DataFrame(list(data), columns=['id', 'substrategy'])
+            # df = pd.read_sql(sql, con).dropna(how='all')
+
+            df.rename({'id': 'fund_id'}, axis=1, inplace=True)
+            manager_info = get_manager(invest_type)
+            df = pd.merge(df, manager_info, how="inner", on='fund_id')
+        return df
 
 
 def resample(df, trading_cal, freq, simple_flag=True):
@@ -301,4 +332,4 @@ if __name__ == '__main__':
     # fund_rank.to_csv("fund_rank.csv", encoding='gbk')
     # df = pd.read_csv('fund_rank.csv')
     # df.to_sql("fund_rank", con, if_exists='replace')
-    con.close()
+    # con.close()