Bitget下载

注册下载Bitget下载，邀请好友，即有机会赢取 3,000 USDT

APP下载   官网注册

陆港通即中国大陆与香港股票市场互联互通机制，由沪港通和深港通组成。沪港通联通沪港股市，于 2014年11月17日开通；深港通联通深港股市，于 2016年12月5日开通。通过陆港通进行的北向交易和南向交易日益活跃，成交额占各自市场(即A股市场和H股市场)百分比在持续地增加。

对于投资者来讲，比较在A股和H股同时上市公司的市场数据可以发现投资机会，同样陆港通也为投资者提供另外一种分析机会。在两地交易的股票价格和成交信息很好获取，一般接口都可以提供，这里我们更加关注的是专用数据，既通过沪(深)股通由香港投资者购买A股股票的成交信息和由港股通的大陆投资者购买H股股票的成交信息，发掘这些指标后面的投资机会。东方财富，同花顺等多个网站提供上述个股成交数据，这里以东方财富为例，沪深港通持股 _ 数据中心 _ 东方财富网

陆港通个股数据，来源：东方财富网数据中心

我们将三个指标：持股数量变化率，持股比例变动额，期间资金净流量分别定义为信号S1，S2，S3。注意三个信号由上表数据计算得到，分别代表一个观察期，比如一周或一个月的期间值，具体计算过程可查看下面代码，或者参考跟随陆港通，A股Alpha与港股Alpha量化策略_CFA Institute China Live

由于网站上只提供最近交易日的个股信息，我将历史数据存到了服务器中，执行下面代码可以直接获取，有条件的同仁也可以自行建立数据库，本数据库的表字段可以在代码块2.3中查询。因为这是个人服务器，算力和带宽有限，只做代码展示用，所以如需下载数据还是建议到专业数据库。

下面内容是我指导学生论文中的节选，他们作为初学者，学习方法和Python编程是主要目的，所以当时只是做了S1，S2，S3三个基本信号，每个信号都需要在一个期间内进行聚合，同时考虑两地有交易时间的差别，这个计算工作量其实并不小。读者可以在此基础上构建更为复杂的因子，包括上图没有使用的其他陆港通指标，甚至和普通个股指标进行结合。

程序库要求

1. pymysql
2. sqlalchemy

以下1-5为函数定义，顺序执行即可

1. 设置数据源

database_ip = '121.41.107.6'account = 'student:VirtualLab123'database_name = 'china_stock_market'

2. 定义函数-数据下载

2.1 获取所有个股的列表

# 香港证券市场个股基本数据def get_hk_basic():    import pymysql    import time    import datetime    from sqlalchemy import create_engine    from sqlalchemy.types import CHAR,INT    pymysql.install_as_MySQLdb()    engine = create_engine('mysql://' + account + '@' + database_ip + ':3306' + '/' + database_name)    import pandas as pd    hk_stock_list = pd.read_sql(sql="SELECT `ts_code` FROM `hk_basic`", con=engine)['ts_code'].values.tolist()    return hk_stock_list# 沪深股市个股基本数据def get_basic():    import pymysql    import time    import datetime    from sqlalchemy import create_engine    from sqlalchemy.types import CHAR,INT    pymysql.install_as_MySQLdb()    engine = create_engine('mysql://' + account + '@' + database_ip + ':3306' + '/' + database_name)    import pandas as pd    stock_list = pd.read_sql(sql="SELECT `ts_code` FROM `stock_basic`", con=engine)['ts_code'].values.tolist()    return stock_list

2.2 股票交易日期

# 香港交易所交易日期列表def get_hk_trade_cal():    import pymysql    import time    import datetime    from sqlalchemy import create_engine    from sqlalchemy.types import CHAR,INT    pymysql.install_as_MySQLdb()    engine = create_engine('mysql://' + account + '@' + database_ip + ':3306' + '/' + database_name)    import pandas as pd    hk_trade_cal = pd.read_sql(sql="SELECT * FROM `hk_trade_cal`", con=engine)    # 增加字段 year    hk_trade_cal["year"] = hk_trade_cal.apply(lambda x:datetime.datetime.strptime(x["cal_date"],"%Y%m%d").isocalendar()[0],axis=1)    # 增加字段 week of year    hk_trade_cal["week"] = hk_trade_cal.apply(lambda x:datetime.datetime.strptime(x["cal_date"],"%Y%m%d").isocalendar()[1],axis=1)    # 增加字段 week_day    hk_trade_cal["week_day"] = hk_trade_cal.apply(lambda x:datetime.datetime.strptime(x["cal_date"],"%Y%m%d").isocalendar()[2],axis=1)    return hk_trade_cal# 上海交易所的交易日历，深交所使用相同日历def get_sh_sz_trade_cal():    import time    import datetime    import pymysql    from sqlalchemy import create_engine    from sqlalchemy.types import CHAR,INT    pymysql.install_as_MySQLdb()    engine = create_engine('mysql://' + account + '@' + database_ip + ':3306' + '/' + database_name)    import pandas as pd    trade_cal = pd.read_sql(sql="SELECT * FROM `trade_cal`", con=engine)    # 增加字段 year    trade_cal["year"] = trade_cal.apply(lambda x:datetime.datetime.strptime(x["cal_date"],"%Y%m%d").isocalendar()[0],axis=1)    # 增加字段 week of year    trade_cal["week"] = trade_cal.apply(lambda x:datetime.datetime.strptime(x["cal_date"],"%Y%m%d").isocalendar()[1],axis=1)    # 增加字段 week_day    trade_cal["week_day"] = trade_cal.apply(lambda x:datetime.datetime.strptime(x["cal_date"],"%Y%m%d").isocalendar()[2],axis=1)    return trade_cal

2.3 个股日交易数据下载

# 下载所有沪深股市个股、所有交易日的数据，field可单选：open,high,low,close,pre_close,change,pct_chg,volume,amountdef get_all_stock_trade(field):     import pymysql    from sqlalchemy import create_engine    from sqlalchemy.types import CHAR,INT    pymysql.install_as_MySQLdb()        engine = create_engine('mysql://' + account + '@' + database_ip + ':3306' + '/' + database_name)    import numpy as np    import pandas as pd    sql = "SELECT `ts_code`,`trade_date`,`"+str(field)+"` FROM `daily` WHERE 1" #SQL query    df = pd.read_sql(sql=sql, con=engine).drop_duplicates() #read data to DataFrame 'df'    dfA= df.pivot(index='trade_date', columns='ts_code', values=str(field))    return dfA# 下载沪深股市个股列表 stock_list 在日期间隔内数据，field可单选：open,high,low,close,pre_close,change,pct_chg,volume,amountdef get_stock_trade(field,stock_list,start_date,end_date):    import pymysql    from sqlalchemy import create_engine    from sqlalchemy.types import CHAR,INT    pymysql.install_as_MySQLdb()    engine = create_engine('mysql://' + account + '@' + database_ip + ':3306' + '/' + database_name)    import numpy as np    import pandas as pd    sql = "SELECT `ts_code`,`trade_date`,`"+str(field)+"` FROM `daily` WHERE `ts_code` in ("+str(stock_list)[1:-1]+") and `trade_date` between "+str(start_date)+" and  "+str(end_date)+""#SQL query    df = pd.read_sql(sql=sql, con=engine).drop_duplicates() #read data to DataFrame 'df'    dfA= df.pivot(index='trade_date', columns='ts_code', values=str(field))    return dfA# 下载沪深股市股票指数 在日期间隔内数据，field可单选：open,high,low,close,pre_close,change,pct_chg,volume,amountdef get_stock_index(field,start_date,end_date):    import pymysql    from sqlalchemy import create_engine    from sqlalchemy.types import CHAR,INT    pymysql.install_as_MySQLdb()    engine = create_engine('mysql://' + account + '@' + database_ip + ':3306' + '/' + database_name)    import numpy as np    import pandas as pd    sql = "SELECT `trade_date`,`"+str(field)+"` FROM `sse_index_daily` WHERE `trade_date` between "+str(start_date)+" and  "+str(end_date)+""#SQL query    df = pd.read_sql(sql=sql, con=engine).drop_duplicates() #read data to DataFrame 'df'    return df# 下载所有香港个股、所有交易日的数据，field可单选：open,high,low,close,pre_close,change,pct_chg,volume,amountdef get_all_hk_stock_trade(field):    import pymysql    from sqlalchemy import create_engine    from sqlalchemy.types import CHAR,INT    pymysql.install_as_MySQLdb()    engine = create_engine('mysql://' + account + '@' + database_ip + ':3306' + '/' + database_name)    import numpy as np    import pandas as pd    sql = "SELECT `ts_code`,`trade_date`,`"+str(field)+"` FROM `hk_daily` WHERE 1" #SQL query    df = pd.read_sql(sql=sql, con=engine).drop_duplicates() #read data to DataFrame 'df'    dfA= df.pivot(index='trade_date', columns='ts_code', values=str(field))    return dfA# 下载香港个股列表 stock_list 在日期间隔内数据，field可单选：open,high,low,close,pre_close,change,pct_chg,volume,amountdef get_hk_stock_trade(field,stock_list,start_date,end_date):    import pymysql    from sqlalchemy import create_engine    from sqlalchemy.types import CHAR,INT    pymysql.install_as_MySQLdb()    engine = create_engine('mysql://' + account + '@' + database_ip + ':3306' + '/' + database_name)    import numpy as np    import pandas as pd    sql = "SELECT `ts_code`,`trade_date`,`"+str(field)+"` FROM `hk_daily` WHERE `ts_code` in ("+str(stock_list)[1:-1]+") and `trade_date` between "+str(start_date)+" and  "+str(end_date)+""#SQL query    df = pd.read_sql(sql=sql, con=engine).drop_duplicates() #read data to DataFrame 'df'    dfA= df.pivot(index='trade_date', columns='ts_code', values=str(field))    return dfA# 下载香港股票指数 在日期间隔内数据，field可单选：open,high,low,close,pre_close,change,pct_chg,volume,amountdef get_hk_stock_index(field,start_date,end_date):    import pymysql    from sqlalchemy import create_engine    from sqlalchemy.types import CHAR,INT    pymysql.install_as_MySQLdb()    engine = create_engine('mysql://' + account + '@' + database_ip + ':3306' + '/' + database_name)    import numpy as np    import pandas as pd    sql = "SELECT `trade_date`,`"+str(field)+"` FROM `hsi_index_daily` WHERE `trade_date` between "+str(start_date)+" and  "+str(end_date)+""#SQL query    df = pd.read_sql(sql=sql, con=engine).drop_duplicates() #read data to DataFrame 'df'    return df

2.4 陆港通专用指标

# 获取陆港通指标# hold_vol 通过陆港通系统交易，持有股票数量,包括南向交易（中国大陆投资者持有香港证券市场股票数量（个股））# hold_ratio 通过陆港通系统交易，持有股票数量占该流通股总数的百分比# 该个股当天的交易量# 该个股当天的成交额def indicator_calc(ts_code,trade_date):    import pymysql    from sqlalchemy import create_engine    from sqlalchemy.types import CHAR,INT    pymysql.install_as_MySQLdb()    engine = create_engine('mysql://' + account + '@' + database_ip + ':3306' + '/' + database_name)    import pandas as pd        market = str(ts_code)[-2:]    ts_code = ts_code    trade_date = trade_date        sql_hold_vol = "SELECT `vol` FROM `sh_sz_hk_hold` WHERE ts_code = "  + "'" + str(ts_code) +"'" + " and trade_date = "+ "'"+ str(trade_date) + "'"    hold_vol = pd.read_sql(sql=sql_hold_vol , con=engine)['vol'][0]    sql_hold_ratio = "SELECT `ratio` FROM `sh_sz_hk_hold` WHERE ts_code = "  + "'" + str(ts_code) +"'" + " and trade_date = "+ "'"+ str(trade_date) + "'"    hold_ratio = pd.read_sql(sql=sql_hold_ratio , con=engine)['ratio'][0]    if market == 'HK':        sql_vol_amount = "SELECT `vol`,`amount` FROM `hk_daily` WHERE ts_code = "  + "'" + str(ts_code) +"'" + " and trade_date = "+ "'"+ str(trade_date) + "'"        trade_vol_amount = pd.read_sql(sql=sql_vol_amount , con=engine)        trade_vol = trade_vol_amount['vol'][0]        trade_amount = trade_vol_amount['amount'][0]    else:        sql_vol_amount = "SELECT `vol`,`amount` FROM `daily` WHERE ts_code = "  + "'" + str(ts_code) +"'" + " and trade_date = "+ "'"+ str(trade_date) + "'"        trade_vol_amount = pd.read_sql(sql=sql_vol_amount , con=engine)        trade_vol = trade_vol_amount['vol'][0]        trade_amount = trade_vol_amount['amount'][0]    return hold_vol, hold_ratio, trade_vol, trade_amount  

3. 定义函数-指标计算

3.1 获取某个市场（HK or SH_SZ) 的股票集合 stock_list 在某个交易日的陆港通专用指标

def indicator_calc_multi_stock(market,stock_list,trade_date):    import pymysql    from sqlalchemy import create_engine    from sqlalchemy.types import CHAR,INT    pymysql.install_as_MySQLdb()    engine = create_engine('mysql://' + account + '@' + database_ip + ':3306' + '/' + database_name)    import pandas as pd        sql_hold_vol = "SELECT `ts_code`,`vol` FROM `sh_sz_hk_hold` WHERE `ts_code` in ("+str(stock_list)[1:-1]+") and trade_date = "+ "'"+ str(trade_date) + "'"    hold_vol = pd.read_sql(sql=sql_hold_vol , con=engine)    hold_vol = hold_vol.set_index('ts_code').astype (float)    sql_hold_ratio = "SELECT `ts_code`,`ratio` FROM `sh_sz_hk_hold` WHERE `ts_code` in ("+str(stock_list)[1:-1]+") and trade_date = "+ "'"+ str(trade_date) + "'"    hold_ratio = pd.read_sql(sql=sql_hold_ratio , con=engine)    hold_ratio = hold_ratio.set_index('ts_code').astype (float)    if market == 'HK':        sql_vol_amount = "SELECT `ts_code`,`vol`,`amount` FROM `hk_daily` WHERE `ts_code` in ("+str(stock_list)[1:-1]+") and trade_date = "+ "'"+ str(trade_date) + "'"        trade_vol_amount = pd.read_sql(sql=sql_vol_amount , con=engine)        trade_vol_amount = trade_vol_amount.set_index('ts_code').astype (float)    else:        sql_vol_amount = "SELECT `ts_code`,`vol`,`amount` FROM `daily` WHERE `ts_code` in ("+str(stock_list)[1:-1]+") and trade_date = "+ "'"+ str(trade_date) + "'"        trade_vol_amount = pd.read_sql(sql=sql_vol_amount , con=engine)        trade_vol_amount = trade_vol_amount.set_index('ts_code').astype (float)    return hold_vol, hold_ratio, trade_vol_amount  

3. 2 信号S1 持股数量变化率

# 计算某个个股在某一年 year, 某一周 week 的选股信号 S1 ，以这一周的周五为观察点def s1_score(ts_code,year,week):    import time    import datetime    from datetime import timedelta    market = str(ts_code)[-2:]    week_day = 5 # 周末观察        # 通过 year，week 推算对应的日期 cal_date 和前一个交易日 pretrade_date    if market == 'HK':        hk_trade_cal = get_hk_trade_cal() # 下载港股日历        cal_date = hk_trade_cal[(hk_trade_cal['year']==year)&(hk_trade_cal['week']==week)&(hk_trade_cal['week_day']==week_day)]['cal_date'].iloc[0]        lastweek = datetime.datetime.strptime(cal_date,"%Y%m%d")-timedelta(days= 7)        pre_week_cal_date = datetime.datetime.strftime(lastweek,"%Y%m%d")    else:        trade_cal = get_sh_sz_trade_cal() # 下载沪深日历        cal_date = trade_cal[(trade_cal['year']==year)&(trade_cal['week']==week)&(trade_cal['week_day']==week_day)]['cal_date'].iloc[0]        lastweek = datetime.datetime.strptime(cal_date,"%Y%m%d")-timedelta(days= 7)        pre_week_cal_date = datetime.datetime.strftime(lastweek,"%Y%m%d")    # 计算该股票 对应日期 cal_date 的指标    hold_vol, hold_ratio, trade_vol, trade_amount = indicator_calc(ts_code,cal_date)    pre_hold_vol, pre_hold_ratio, pre_trade_vol, pre_trade_amount = indicator_calc(ts_code,pre_week_cal_date)    return float(hold_vol)/float(pre_hold_vol)-1# 计算香港市场或沪深股市的个股列表在某一年 year, 某一周 week 的选股信号 S1，以这一周的周五为观察点def s1_score_multi_stock(market,stock_list,year,week):    import time    import datetime    from datetime import timedelta    week_day = 5 # 周末观察        # 通过 year，week 推算对应的日期 cal_date 和前一个交易日 pretrade_date    if market == 'HK':        hk_trade_cal = get_hk_trade_cal() # 下载港股日历        cal_date = hk_trade_cal[(hk_trade_cal['year']==year)&(hk_trade_cal['week']==week)&(hk_trade_cal['week_day']==week_day)]['cal_date'].iloc[0]        lastweek = datetime.datetime.strptime(cal_date,"%Y%m%d")-timedelta(days= 7)        pre_week_cal_date = datetime.datetime.strftime(lastweek,"%Y%m%d")    else:        trade_cal = get_sh_sz_trade_cal() # 下载沪深日历        cal_date = trade_cal[(trade_cal['year']==year)&(trade_cal['week']==week)&(trade_cal['week_day']==week_day)]['cal_date'].iloc[0]        lastweek = datetime.datetime.strptime(cal_date,"%Y%m%d")-timedelta(days= 7)        pre_week_cal_date = datetime.datetime.strftime(lastweek,"%Y%m%d")    # 计算该股票 对应日期 cal_date 的指标    hold_vol, hold_ratio, trade_vol_amount = indicator_calc_multi_stock(market,stock_list,cal_date)    pre_hold_vol, pre_hold_ratio, pre_trade_vol_amount = indicator_calc_multi_stock(market,stock_list,pre_week_cal_date)    return hold_vol/pre_hold_vol-1

3.3 信号S2：持股比例变动额

# 计算某个个股在某一年 year, 某一周 week 的选股信号 S2 ，以这一周的周五为观察点def s2_score(ts_code,year,week):    import time    import datetime    from datetime import timedelta    market = str(ts_code)[-2:]    week_day = 5 # 周末观察        # 通过 year，week 推算对应的日期 cal_date 和前一个交易日 pretrade_date    if market == 'HK':        hk_trade_cal = get_hk_trade_cal() # 下载港股日历        cal_date = hk_trade_cal[(hk_trade_cal['year']==year)&(hk_trade_cal['week']==week)&(hk_trade_cal['week_day']==week_day)]['cal_date'].iloc[0]        lastweek = datetime.datetime.strptime(cal_date,"%Y%m%d")-timedelta(days= 7)        pre_week_cal_date = datetime.datetime.strftime(lastweek,"%Y%m%d")    else:        trade_cal = get_sh_sz_trade_cal() # 下载沪深日历        cal_date = trade_cal[(trade_cal['year']==year)&(trade_cal['week']==week)&(trade_cal['week_day']==week_day)]['cal_date'].iloc[0]        lastweek = datetime.datetime.strptime(cal_date,"%Y%m%d")-timedelta(days= 7)        pre_week_cal_date = datetime.datetime.strftime(lastweek,"%Y%m%d")    # 计算该股票 对应日期 cal_date 的指标    hold_vol, hold_ratio, trade_vol, trade_amount = indicator_calc(ts_code,cal_date)    pre_hold_vol, pre_hold_ratio, pre_trade_vol, pre_trade_amount = indicator_calc(ts_code,pre_week_cal_date)    return float(hold_ratio)-float(pre_hold_ratio)# 计算香港市场或沪深股市的个股列表在某一年 year, 某一周 week 的选股信号 S2，以这一周的周五为观察点def s2_score_multi_stock(market,stock_list,year,week):    import time    import datetime    from datetime import timedelta    week_day = 5 # 周末观察        # 通过 year，week 推算对应的日期 cal_date 和前一个交易日 pretrade_date    if market == 'HK':        hk_trade_cal = get_hk_trade_cal() # 下载港股日历        cal_date = hk_trade_cal[(hk_trade_cal['year']==year)&(hk_trade_cal['week']==week)&(hk_trade_cal['week_day']==week_day)]['cal_date'].iloc[0]        lastweek = datetime.datetime.strptime(cal_date,"%Y%m%d")-timedelta(days= 7)        pre_week_cal_date = datetime.datetime.strftime(lastweek,"%Y%m%d")    else:        trade_cal = get_sh_sz_trade_cal() # 下载沪深日历        cal_date = trade_cal[(trade_cal['year']==year)&(trade_cal['week']==week)&(trade_cal['week_day']==week_day)]['cal_date'].iloc[0]        lastweek = datetime.datetime.strptime(cal_date,"%Y%m%d")-timedelta(days= 7)        pre_week_cal_date = datetime.datetime.strftime(lastweek,"%Y%m%d")    # 计算该股票 对应日期 cal_date 的指标    hold_vol, hold_ratio, trade_vol_amount = indicator_calc_multi_stock(market,stock_list,cal_date)    pre_hold_vol, pre_hold_ratio, pre_trade_vol_amount = indicator_calc_multi_stock(market,stock_list,pre_week_cal_date)    return hold_ratio-pre_hold_ratio

3.4 信号S3: 期间资金净流量

# 计算某个个股在某一年 year, 某一周 week 的选股信号 S3 ，以这一周的周五为观察点,计算过去n天总的资金净流量(day)，# 每一天的资金净流量 = （当天持股数量 - 前一天持股数量）* VWAP (成交量加权平均价 = amount / volume)def s3_score(ts_code,year,week,day):    import pandas as pd    import time    import datetime    from datetime import timedelta    market = str(ts_code)[-2:]    week_day = 5 # 周末观察        # 通过 year，week 推算对应的日期 cal_date 和前一个交易日 pretrade_date    if market == 'HK':        hk_trade_cal = get_hk_trade_cal() # 下载港股日历        cal_date = hk_trade_cal[(hk_trade_cal['year']==year)&(hk_trade_cal['week']==week)&(hk_trade_cal['week_day']==week_day)]['cal_date'].iloc[0]        cal_date_object = datetime.datetime.strptime(cal_date,"%Y%m%d") # 日期对象        pre_week_object = cal_date_object-timedelta(days= day) #日期对象        pre_week_cal_date = datetime.datetime.strftime(pre_week_object,"%Y%m%d")#日期对象转化为字符串日期    else:        trade_cal = get_sh_sz_trade_cal() # 下载沪深日历        cal_date = trade_cal[(trade_cal['year']==year)&(trade_cal['week']==week)&(trade_cal['week_day']==week_day)]['cal_date'].iloc[0]        cal_date_object = datetime.datetime.strptime(cal_date,"%Y%m%d") # 日期对象        pre_week_object = cal_date_object-timedelta(days= day) #日期对象        pre_week_cal_date = datetime.datetime.strftime(pre_week_object,"%Y%m%d")#日期对象转化为字符串日期        # 计算该股票 对应日期 pre_week_cal_date, cal_date 之间的指标求和    hold_vol_list =[]    trade_price_list = []    while pre_week_object < cal_date_object:        date = datetime.datetime.strftime(pre_week_object,"%Y%m%d")        try:            hold_vol, hold_ratio, trade_vol, trade_amount = indicator_calc(ts_code,date)            hold_vol_list.append([date,hold_vol])            trade_price_list.append([date,float(trade_amount)/float(trade_vol)])            pre_week_object = pre_week_object + timedelta(days=1)            print (date)        except:            pre_week_object = pre_week_object + timedelta(days=1)            print (pre_week_object)            continue    hold_vol_list_df = pd.DataFrame(hold_vol_list,columns=['date','hold_vol'])    hold_vol_list_df.index = hold_vol_list_df['date']    hold_vol_list_df_shift = hold_vol_list_df.shift(1) # 滞后一期，index不会随着滞后    del hold_vol_list_df_shift['date']    del hold_vol_list_df['date']        hold_vol_change_list_df = hold_vol_list_df.astype('float') - hold_vol_list_df_shift.astype('float')           trade_price_list_df = pd.DataFrame(trade_price_list,columns=['date','hold_vol']) # 只有列名一致才能计算，其实这个是vwap    trade_price_list_df.index = trade_price_list_df['date']    s3_t = hold_vol_change_list_df * trade_price_list_df    s3 = s3_t.sum()        return s3[1]# 计算香港市场或沪深股市的个股列表在某一年 year, 某一周 week 的选股信号 S3，以这一周的周五为观察点def s3_score_multi_stock(market,stock_list,year,week,day):    import pandas as pd    import time    import datetime    from datetime import timedelta    week_day = 5 # 周末观察        # 通过 year，week 推算对应的日期 cal_date 和前一个交易日 pretrade_date    if market == 'HK':        hk_trade_cal = get_hk_trade_cal() # 下载港股日历        cal_date = hk_trade_cal[(hk_trade_cal['year']==year)&(hk_trade_cal['week']==week)&(hk_trade_cal['week_day']==week_day)]['cal_date'].iloc[0]        cal_date_object = datetime.datetime.strptime(cal_date,"%Y%m%d") # 日期对象        pre_week_object = cal_date_object-timedelta(days= day) #日期对象        pre_week_cal_date = datetime.datetime.strftime(pre_week_object,"%Y%m%d")#日期对象转化为字符串日期    else:        trade_cal = get_sh_sz_trade_cal() # 下载沪深日历        cal_date = trade_cal[(trade_cal['year']==year)&(trade_cal['week']==week)&(trade_cal['week_day']==week_day)]['cal_date'].iloc[0]        cal_date_object = datetime.datetime.strptime(cal_date,"%Y%m%d") # 日期对象        pre_week_object = cal_date_object-timedelta(days= day) #日期对象        pre_week_cal_date = datetime.datetime.strftime(pre_week_object,"%Y%m%d")#日期对象转化为字符串日期        # 计算股票列表 stock_list 对应日期 pre_week_cal_date, cal_date 之间的指标求和    hold_vol_df = pd.DataFrame()    trade_price_df = pd.DataFrame()    while pre_week_object < cal_date_object:        date = datetime.datetime.strftime(pre_week_object,"%Y%m%d")        try:            hold_vol, hold_ratio, trade_vol_amount = indicator_calc_multi_stock(market,stock_list,date)            hold_vol = hold_vol.T            hold_vol['date'] = date            hold_vol_df = hold_vol_df.append(hold_vol)                        trade_vol_amount['vwap'] = (trade_vol_amount['amount'].astype('float')/trade_vol_amount['vol']).astype('float')            del trade_vol_amount['amount']            del trade_vol_amount['vol']            trade_price = trade_vol_amount.T            trade_price['date'] =date            trade_price_df = trade_price_df.append(trade_price)            pre_week_object = pre_week_object + timedelta(days=1)            print (date)        except:            pre_week_object = pre_week_object + timedelta(days=1)            print (pre_week_object)            continue    hold_vol_df = hold_vol_df.set_index(['date'])    trade_price_df = trade_price_df.set_index(['date'])        # 计算该股票 对应日期 cal_date 的指标    s = (hold_vol_df-hold_vol_df.shift(1))*trade_price_df    s3 = pd.DataFrame(s.sum(axis=0))    s3.columns = ['netflow']    return s3

4. 定义选股过程

4.1 给定信号，时间区间，计算出所有股票的得分

def  selection_stock(market,signal,start_year,end_year,start_week,end_week,day):    import datetime    start_time = datetime.datetime.now()    if market == 'HK':        stock_list = get_hk_basic()    elif market == 'SZ':        stock_list = get_basic()    elif market == 'SH':        stock_list = get_basic()    import pandas as pd    score_list = []    if signal == 's1':        for year in range(start_year,end_year):            print(year)            for week in range(start_week,end_week):                print("week is:",week,datetime.datetime.now())                try:                    s = s1_score_multi_stock(market,stock_list,year,week)                    score_list.append([year,week,s])                except:                    continue        if signal == 's2':        for year in range(start_year,end_year):            print(year)            for week in range(start_week,end_week):                print("week is:",week,datetime.datetime.now())                try:                    s = s2_score_multi_stock(market,stock_list,year,week)                    score_list.append([year,week,s])                except:                    continue        if signal == 's3':        for year in range(start_year,end_year):            print(year)            for week in range(start_week,end_week):                print("week is:",week,datetime.datetime.now())                try:                    s = s3_score_multi_stock(market,stock_list,year,week,day)                    score_list.append([year,week,s])                except:                    continue        finish_time = datetime.datetime.now()    print (finish_time - start_time)    return score_list

4.2 根据得分对股票进行排序

def ranked_stock(score_list,percent):    ranked_stock_list = []    for stock_list in score_list:        year = stock_list[0]        week = stock_list[1]        variable = stock_list[2].columns[0]        ranked_list = stock_list[2].sort_values(by=variable, ascending = False)        top_number = int(len(ranked_list)*percent)        top_stock_in_list = ranked_list[:top_number].index.to_list()        ranked_stock_list.append([year,week,top_stock_in_list])    return ranked_stock_list

5. 选股策略的收益率回测

# 计算给定市场market，股票stock_list在给定年year，周week，投资一定期限weeks_invest的收益def backtest_multi_stock(market,stock_list,year,week,weeks_invest):    #计算选出的一组股票，在year, week 投资 weeks_invest 周的平均收益    import time    import datetime    from datetime import timedelta    week_day = 5 # 周末观察    # weeks_invest 为1,2,...,n    # 通过 year，week 推算对应的日期 cal_date的下周一作为买入，下周五或者下下周五卖出    if market == 'HK':        hk_trade_cal = get_hk_trade_cal() # 下载港股日历        cal_date = hk_trade_cal[(hk_trade_cal['year']==year)&(hk_trade_cal['week']==week)&(hk_trade_cal['week_day']==week_day)]['cal_date'].iloc[0]        buy_date_object = datetime.datetime.strptime(cal_date,"%Y%m%d")+timedelta(days= 3)        buy_date = datetime.datetime.strftime(buy_date_object,"%Y%m%d")        sell_date_object = datetime.datetime.strptime(cal_date,"%Y%m%d")+timedelta(days= 7*weeks_invest)        sell_date = datetime.datetime.strftime(sell_date_object,"%Y%m%d")        buy_price = get_hk_stock_trade('close',stock_list,buy_date,buy_date)        sell_price = get_hk_stock_trade('close',stock_list,sell_date,sell_date)        buy_index = get_hk_stock_index('close',buy_date,buy_date)['close'][0]        sell_index = get_hk_stock_index('close',sell_date,sell_date)['close'][0]    else:        trade_cal = get_sh_sz_trade_cal() # 下载沪深日历        cal_date = trade_cal[(trade_cal['year']==year)&(trade_cal['week']==week)&(trade_cal['week_day']==week_day)]['cal_date'].iloc[0]        buy_date_object = datetime.datetime.strptime(cal_date,"%Y%m%d")+timedelta(days= 3)        buy_date = datetime.datetime.strftime(buy_date_object,"%Y%m%d")        sell_date_object = datetime.datetime.strptime(cal_date,"%Y%m%d")+timedelta(days= 7*weeks_invest)        sell_date = datetime.datetime.strftime(sell_date_object,"%Y%m%d")        buy_price = get_stock_trade('close',stock_list,buy_date,buy_date)        sell_price = get_stock_trade('close',stock_list,sell_date,sell_date)        buy_index = get_stock_index('close',buy_date,buy_date)['close'][0]        sell_index = get_stock_index('close',sell_date,sell_date)['close'][0]        buy_price = buy_price.T    buy_price.columns = ['close']    sell_price = sell_price.T    sell_price.columns = ['close']        return ((sell_price/buy_price).mean()-1)[0],sell_index/buy_index-1

# 将上述结果进行汇总def main_cal(market,ranked_stock_list,weeks_invest):    test_result = []    for stock_list in ranked_stock_list:        try:            year = stock_list[0]            week = stock_list[1]            s1,s2 = backtest_multi_stock(market,stock_list[2],year,week,weeks_invest)            test_result.append([year,week,s1,s2,s1-s2])        except:            continue    return test_result

以下参数需要自行设定，并可以反复尝试

1. 参数设定

market = 'HK' # 可选SH,SZ(SH,SZ结果一致，都代表北向交易）HK,代表南向交易signal = 's2' # 可选s1,s2,s3start_year = 2017 # 可选2017-2021，策略回测起始年，这里建议不改动end_year = 2028 # 可选2017-2021，策略回测结束年，这里建议不改动start_week = 1 # 可选1-53，策略回测的起始周，这里建议不改动end_week = 53 # 可选1-53，策略回测的结束周，这里建议不改动percent = 0.2 # 选取排名在前百分之多少的股票买入day = 5 # 信号3需要指定累计过去多少天的资金净流入weeks_invest = 1 # 按照选定策略投资多少天，建议选择1-20

2. 根据设定参数计算因子得分

stock_list = selection_stock(market,signal,start_year,end_year,start_week,end_week,day)

3. 根据得分排名

ranked_stock_list = ranked_stock(stock_list,percent)

4. 进行回测

main_result = main_cal(market,ranked_stock_list,weeks_invest)

5. 把回测结果进行整理

import pandas as pdmain_result_df = pd.DataFrame(main_result,columns=['year','week','return','stock_index','access_return'])# 整理绝对收益return_pivot = main_result_df.pivot(index='week',columns='year',values='return').reset_index(drop=True)# 整理大盘收益stock_index = main_result_df.pivot(index='week',columns='year',values='stock_index').reset_index(drop=True)# 整理超额收益access_return_pivot = main_result_df.pivot(index='week',columns='year',values='access_return').reset_index(drop=True)

6. 结果输出和保存

# 投资策略每周的绝对收益，第一句代码会在本地生成csv文件return_pivot.fillna(0).to_csv('return'+str(market)+str(signal)+'.csv')return_pivot.fillna(0).style.background_gradient(cmap='bwr',vmin=-0.1,vmax=0.1)# cmap：颜色; # axis：映射参考，0为行，1以列

# 投资策略每周超额收益，第一句代码会在本地生成csv文件access_return_pivot.fillna(0).to_csv('access_return'+str(market)+str(signal)+'.csv')access_return_pivot.fillna(0).style.background_gradient(cmap='bwr',vmin=-0.1,vmax=0.1)# cmap：颜色; # axis：映射参考，0为行，1以列

# 计算累计收益并且绘图import pandas as pdreturn_acc = pd.melt(return_pivot.fillna(0))+1index_acc = pd.melt(stock_index.fillna(0))+1compare = pd.concat([return_acc['value'],index_acc['value']],axis=1)compare.columns = ['return_acc','index_acc']final_compare =pd.DataFrame(data=[[1,1]],columns=['return_acc','index_acc'])for i in range(len(compare)):    final_compare = final_compare.append(final_compare.iloc[-1,:]*compare.iloc[i,:],ignore_index=True)final_compare.plot()

相关内容