接入tamp产品池排名

app/utils/fund_rank.py

@@ -8,7 +8,7 @@
 import logging
 logging.basicConfig(level=logging.INFO)
 
-from app.api.engine import tamp_fund_engine, TAMP_SQL
+from app.api.engine import tamp_fund_engine, TAMP_SQL, tamp_product_engine
 from app.utils.week_evaluation import *
 
 
@@ -32,39 +32,70 @@ def get_nav(fund, start_date, rollback=False, invest_type='public'):
     Returns:df[DataFrame]: 索引为净值公布日， 列为复权净值的净值表; 查询失败则返回None
 
     """
-    with TAMP_SQL(tamp_fund_engine) as tamp_product:
-        tamp_product_session = tamp_product.session
+    with TAMP_SQL(tamp_fund_engine) as tamp_fund:
+        tamp_fund_session = tamp_fund.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)
+            cur = tamp_fund_session.execute(sql)
             data = cur.fetchall()
-            df = pd.DataFrame(list(data), columns=['fund_id', 'end_date', 'adj_nav']).dropna(how='any')
-            df.rename({'ts_code': 'fund_id'}, axis=1, inplace=True)
+            df1 = pd.DataFrame(list(data), columns=['fund_id', 'end_date', 'adj_nav']).dropna(how='any')
+            df1.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_fund_session.execute(sql)
+            data = cur.fetchall()
+            df1 = pd.DataFrame(data, columns=['fund_id', 'price_date', 'cumulative_nav']).dropna(how='any')
+            df1.rename({'price_date': 'end_date', 'cumulative_nav': 'adj_nav'}, axis=1, inplace=True)
+
+        # if df1['adj_nav'].count() == 0:
+        #     logging.log(logging.ERROR, "CAN NOT FIND {}".format(fund))
+        #     return None
+
+        df1['end_date'] = pd.to_datetime(df1['end_date'])
+
+        if rollback and df1['end_date'].min() < start_date < df1['end_date'].max():
+            while start_date not in list(df1['end_date']):
+                start_date -= datetime.timedelta(days=1)
+
+        df1 = df1[df1['end_date'] >= start_date]
+        df1.drop_duplicates(subset='end_date', inplace=True, keep='first')
+        # df1.set_index('end_date', inplace=True)
+        # df1.sort_index(inplace=True, ascending=True)
+
+    with TAMP_SQL(tamp_product_engine) as tamp_product:
+        tamp_product_session = tamp_product.session
+        if invest_type == "private":
             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)
+            df2 = pd.DataFrame(data, columns=['fund_id', 'price_date', 'cumulative_nav']).dropna(how='any')
+            df2.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
+        # if df2['adj_nav'].count() == 0:
+        #     logging.log(logging.ERROR, "CAN NOT FIND {}".format(fund))
+        #     return None
 
-        df['end_date'] = pd.to_datetime(df['end_date'])
+        df2['end_date'] = pd.to_datetime(df2['end_date'])
 
-        if rollback and df['end_date'].min() < start_date < df['end_date'].max():
-            while start_date not in list(df['end_date']):
+        if rollback and df2['end_date'].min() < start_date < df2['end_date'].max():
+            while start_date not in list(df2['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
+        df2 = df2[df2['end_date'] >= start_date]
+        df2.drop_duplicates(subset='end_date', inplace=True, keep='first')
+        # df2.set_index('end_date', inplace=True)
+        # df2.sort_index(inplace=True, ascending=True)
+
+    df = pd.concat([df1, df2])
+    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):
@@ -101,10 +132,10 @@ def get_trade_cal():
     Returns:df[DataFrame]: 索引为交易日， 列为交易日的上交所交易日历表
 
     """
-    with TAMP_SQL(tamp_fund_engine) as tamp_product:
-        tamp_product_session = tamp_product.session
+    with TAMP_SQL(tamp_fund_engine) as tamp_fund:
+        tamp_fund_session = tamp_fund.session
         sql = 'SELECT cal_date FROM stock_trade_cal WHERE is_open=1'
-        cur = tamp_product_session.execute(sql)
+        cur = tamp_fund_session.execute(sql)
         data = cur.fetchall()
         df = pd.DataFrame(list(data), columns=['cal_date']).dropna(how='all')
         # df = pd.read_sql(sql, con)
@@ -122,21 +153,31 @@ def get_manager(invest_type):
     Returns:
 
     """
-    with TAMP_SQL(tamp_fund_engine) as tamp_product:
-        tamp_product_session = tamp_product.session
+    with TAMP_SQL(tamp_fund_engine) as tamp_fund:
+        tamp_fund_session = tamp_fund.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)
+            cur = tamp_fund_session.execute(sql)
             data = cur.fetchall()
-            df = pd.DataFrame(list(data), columns=['ts_code', 'name'])
+            df1 = pd.DataFrame(list(data), columns=['ts_code', 'name'])
         else:
-            sql = 'SELECT fund_id, fund_manager_id FROM fund_manager_mapping'
+            sql = 'SELECT fund_id, fund_manager_id, management_start_date FROM fund_manager_mapping'
             # df = pd.read_sql(sql, con)
+            cur = tamp_fund_session.execute(sql)
+            data = cur.fetchall()
+            df1 = pd.DataFrame(list(data), columns=['fund_id', 'fund_manager_id', 'management_start_date'])
+
+    with TAMP_SQL(tamp_product_engine) as tamp_product:
+        tamp_product_session = tamp_product.session
+        if invest_type == 'private':
+            sql = 'SELECT fund_id, fund_manager_id, management_start_date FROM fund_manager_mapping'
             cur = tamp_product_session.execute(sql)
             data = cur.fetchall()
-            df = pd.DataFrame(list(data), columns=['fund_id', 'fund_manager_id'])
-        return df
+            df2 = pd.DataFrame(list(data), columns=['fund_id', 'fund_manager_id', 'management_start_date'])
+
+    df = pd.concat([df1, df2]).drop_duplicates()
+    return df
 
 
 def get_fund_info(end_date, invest_type):
@@ -149,33 +190,49 @@ def get_fund_info(end_date, invest_type):
     Returns:
         [type]: [description]
     """
-    with TAMP_SQL(tamp_fund_engine) as tamp_product:
-        tamp_product_session = tamp_product.session
+    with TAMP_SQL(tamp_fund_engine) as tamp_fund:
+        tamp_fund_session = tamp_fund.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)
+            cur = tamp_fund_session.execute(sql)
             data = cur.fetchall()
 
-            df = pd.DataFrame(list(data), columns=['ts_code', 'fund_type', 'management'])
-            manager_info = get_manager(invest_type)
+            df1 = pd.DataFrame(list(data), columns=['fund_id', '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')
+            # df1.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"
+            cur = tamp_fund_session.execute(sql)
+            data = cur.fetchall()
+            df1 = pd.DataFrame(list(data), columns=['fund_id', 'substrategy'])
+            # df = pd.read_sql(sql, con).dropna(how='all')
+
+            # df1.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')
 
+    with TAMP_SQL(tamp_product_engine) as tamp_product:
+        tamp_product_session = tamp_product.session
+        if invest_type == "private":
             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'])
+            df2 = pd.DataFrame(list(data), columns=['fund_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
+            # df2.rename({'id': 'fund_id'}, axis=1, inplace=True)
+            # manager_info = get_manager(invest_type)
+
+    manager_info = get_manager(invest_type)
+    df = pd.concat([df1, df2]).drop_duplicates()
+    df = pd.merge(df, manager_info, how="inner", on='fund_id')
+    return df
 
 
 def resample(df, trading_cal, freq, simple_flag=True):
@@ -276,11 +333,12 @@ def fund_rank(start_date, end_date, invest_type='private'):
                     continue
                 n = get_frequency(df)
             except Exception as e:
-                # logging.log(logging.ERROR, repr(e))
+                logging.log(logging.ERROR, repr(e))
                 logging.log(logging.INFO, 'Skipped {}'.format(fund))
+                skipped_funds.append(fund)
                 continue
 
-            df = resample(df, trading_cal, n)
+            df = resample(df, trading_cal, n, simple_flag=False)
 
             try:
                 _ = get_frequency(df)
@@ -302,7 +360,8 @@ def fund_rank(start_date, end_date, invest_type='private'):
             down_risk = downside_risk(sim_return, bank_rate=0.015, n=n)
             sor_ratio = sortino_ratio(ex_return, down_risk, n)
 
-            manager = fund_info[fund_info['fund_id'] == fund]['fund_manager_id'].values
+            manager = fund_info[fund_info['fund_id'] == fund].groupby("management_start_date").last()
+            manager = manager['fund_manager_id'].values
             # management = fund_info[fund_info['fund_id'] == fund]['management'].values
 
             row = pd.Series([fund, rng_return, ann_return, drawdown, shp_ratio,
@@ -324,8 +383,8 @@ def fund_rank(start_date, end_date, invest_type='private'):
 if __name__ == '__main__':
     end_date = datetime.datetime.now() - datetime.timedelta(days=1)
     start_date = cal_date(end_date, 'Y', 1)
-    fund_rank = fund_rank(start_date, end_date, False)
-    # fund_rank.to_csv("fund_rank.csv", encoding='gbk')
-    # df = pd.read_csv('fund_rank.csv')
+    fund_rank = fund_rank(start_date, end_date, invest_type="private")
+    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()