WorldQuant Alpha101因子 附录四：对Alpha101因子的因子分析示例（以Alpha#100为例）
由qxiao创建，最终由qxiao 被浏览 363 用户
Step 1 导入相关包
import pandas as pd
import numpy as np
import warnings
import empyrical
import dai
import bigcharts
warnings.filterwarnings('ignore')
from biglearning.api import tools as T
print('导入包完成!')
Step 2 读取因子数据、设置因子分析参数并进行因子数据预处理
params = {'group_num':10, 'factor_field':'alpha_6100', 'instruments':'全市场', 'factor_direction':1, 'benchmark':'中证500', 'data_process':True} # instruments支持选项：沪深300、中证500、中证1000、全市场；benchmark支持的选项：沪深300、中证500、中证1000
sql = """
WITH industry_data AS (
SELECT instrument, date, industry_level1_code
FROM cn_stock_industry_component
WHERE date >= '2015-01-01' AND industry = 'sw2021'
),
windows AS (
SELECT instrument, date, ((close - low)-(high - close))/(high - low)*volume AS pre_1,
close, m_avg(amount, 20) AS adv20, volume, amount,
ROW_NUMBER() OVER (PARTITION BY instrument ORDER BY date DESC) AS rn,
m_min(close, 30) AS min_close
FROM cn_stock_bar1d
WHERE date >= '2015-01-01'
),
windows2 AS (
SELECT w1.instrument, w1.date, w1.pre_1, w1.close, w1.adv20, w1.volume, MIN(w2.rn) AS rn2, rn2 - w1.rn AS ts_argmin, w1.amount
FROM windows w1
LEFT JOIN windows w2 ON w1.instrument = w2.instrument AND w1.min_close = w2.close
WHERE w1.rn <= w2.rn AND w2.rn < w1.rn + 30
GROUP BY w1.instrument, w1.date, w1.pre_1, w1.rn, w1.close, w1.adv20, w1.volume, w1.amount
),
merged_data AS (
SELECT b.instrument, b.date,
b.pre_1 - AVG(b.pre_1) OVER (PARTITION BY a.industry_level1_code, a.date) AS indus_pre_1,
indus_pre_1 - AVG(indus_pre_1) OVER (PARTITION BY a.industry_level1_code, a.date) AS indus_pre,
1.5 * ABS(indus_pre)/SUM(ABS(indus_pre)) OVER (PARTITION BY b.date) AS factor_1,
b.close, b.adv20, b.volume, b.ts_argmin, b.amount
FROM industry_data a
JOIN windows2 b ON
a.instrument = b.instrument AND a.date = b.date
),
merged_data2 AS (
SELECT instrument, date, factor_1, amount,
m_corr(close, pct_rank_by(date, adv20), 5) - pct_rank_by(date, ts_argmin) AS indus_pre_2,
volume, adv20
FROM merged_data
)
SELECT b.instrument, b.date, b.amount,
b.indus_pre_2 - AVG(b.indus_pre_2) OVER (PARTITION BY a.industry_level1_code, a.date) AS indus_pre2,
ABS(indus_pre2)/SUM(ABS(indus_pre2)) OVER (PARTITION BY b.date) AS factor_2,
0 - ((b.factor_1 - factor_2) * (b.volume/b.adv20)) AS alpha_6100
FROM industry_data a
JOIN merged_data2 b ON
a.instrument = b.instrument AND a.date = b.date
ORDER BY b.date, b.instrument;
"""
factor_data = dai.query(sql).df()
# 因子数据处理
factor_data.dropna(subset=[params['factor_field']], inplace=True)
factor_data = factor_data[['instrument', 'date', 'amount', params['factor_field']]]
# 因子数据更多的预处理，包括去除ST、新股、北交所的股票
def factor_data_filter(factor_data):
columns = factor_data.columns
start_date = factor_data.date.min().strftime('%Y-%m-%d')
end_date = factor_data.date.max().strftime('%Y-%m-%d')
factor_data['instrument'] = factor_data['instrument'].apply(lambda x:x[:9])
base_info_df = dai.query("select date, instrument, st_status ,trading_days from cn_stock_factors_base where date >= '%s' and date <= '%s'"%(start_date, end_date)).df()
factor_data = pd.merge(factor_data, base_info_df, how='left', on=['date', 'instrument'])
factor_data = factor_data[(factor_data['st_status'] == 0) & (factor_data['trading_days']> 252)] # 去除st 和上市不足一年的票
factor_data= factor_data[factor_data.instrument.apply(lambda x: True if x.endswith('SH') or x.endswith('SZ') else False)] # 去除北交所的票
factor_data = factor_data[factor_data['amount'] > 0 ] # 去除停牌期间的数据
factor_data.replace([np.inf, -np.inf], np.nan, inplace=True) # 将 inf 替换为 NaN
# 删除包含 NaN 的行
factor_data.dropna(inplace=True)
return factor_data[columns]
factor_data = factor_data_filter(factor_data)
print('因子数据过滤完成')
*因子分析框架展示
### 因子分析工具
class AlphaMiner(object):
def __init__(self, params, factor_data):
# params: 字典格式。 形如 {'group_num':10, 'factor_field':'hf_close_netinflow_rate_small_order_act', 'instruments':'中证500', 'factor_direction':-1, 'benchmark':'中证500'}
# group_num:分组数量 参数类型：int
# factor_field:因子在表中所对应的字段名称 参数类型：str
# instruments:标的池,支持选项：沪深300、中证500、中证1000、全市场 参数类型：str
# factor_direction：因子方向，字符串格式，取值为1、-1。1表示因子方向为正，因子值越大越好，-1表示因子值为负，因子值越小越好。 参数类型：int
# benchmark:基准对比指数，支持选项：沪深300、中证500、中证1000 参数类型：str
# factor_data:pandas.DataFrame格式,形如
# instrument date hf_fz_ykws
# 0 000001.SZ 2017-01-03 1.564644
# 1 000001.SZ 2017-01-04 1.521567
# 2 000001.SZ 2017-01-05 1.519973
# 3 000001.SZ 2017-01-06 1.553225
# 4 000001.SZ 2017-01-09 1.367971
# 其中, instrument:str ,以股票代码+.sh（沪市） +.SZ（深市）
# date:datetime64
# hf_fz_ykws:float64
self.params = params
self.top_n_ins = 5 # 默认5只
self.factor_data = factor_data.rename(columns={self.params['factor_field']:'factor'})
self.factor_data['factor'] *= self.params['factor_direction']
if self.params['data_process'] == True:
self.factor_data = self.factor_data_process('factor')
print("数据预处理完成")
# 检查因子数据格式
try:
self.check_data_format(self.factor_data)
print("数据格式检查通过")
except ValueError as e:
print("数据格式检查失败：" + str(e))
# 进行数据池过滤
self.stock_pool_filter()
self.start_date = self.factor_data.date.min().strftime('%Y-%m-%d')
self.end_date = self.factor_data.date.max().strftime('%Y-%m-%d')
self.price_data = self.get_daily_ret(self.start_date, self.end_date) # 日收益率数据
print('个股日收益率计算完成')
self.merge_data = pd.merge(self.factor_data.sort_values(['date', 'instrument']), \
self.price_data.sort_values(['date', 'instrument']), on=['date','instrument'], how='left')
self.group_data = self.get_group_data() # 分组数据
print('因子分组完成')
self.bm_ret = self.get_bm_ret(self.params['benchmark'])
print('基准日收益率计算完成')
self.group_cumret = self.get_group_cumret() # 分组累积收益率
print('分组收益率计算完成')
self.whole_perf = self.get_whole_perf() # 整体绩效指标
print('整体绩效计算完成')
self.yearly_perf = self.get_yearly_perf() # 按年度绩效指标
print('年度绩效计算完成')
self.ic = self.get_IC_data('all') # ic指标
print('IC计算完成')
def factor_data_process(self, col):
"""因子数据预处理函数，包括去极值、标准化、中性化"""
def zscore(df, train_col):
from sklearn.preprocessing import StandardScaler
scaler = StandardScaler()
df[train_col] = scaler.fit_transform(df[train_col])
return df
def remove_extreme_and_cut_zscore(df, col):
col_list = [col]
for fac in col_list:
n = 3
mean = df[fac].mean() # 计算因子值的均值
std = df[fac].std() # 计算因子值的标准差
lower_bound = mean - n * std # 计算下边界
upper_bound = mean + n * std # 计算上边界
df.loc[df[fac]<lower_bound, fac] = lower_bound
df.loc[df[fac]>upper_bound, fac] = upper_bound
df = zscore(df, col_list)
return df
factor_data = self.factor_data.groupby('date').apply(remove_extreme_and_cut_zscore, col=col)
factor_data = factor_data.fillna(0) # 用0进行填充
start_date = factor_data.date.min().strftime("%Y-%m-%d")
end_date = factor_data.date.max().strftime("%Y-%m-%d")
sql = """
SELECT date, instrument, industry_level1_code
FROM cn_stock_industry_component
WHERE date >= '{0}' and date <= '{1}' and industry =='sw2021'
ORDER BY instrument, date;
""".format(start_date, end_date)
df_industry = dai.query(sql).df()
df_industry = df_industry.dropna()
factor_data = factor_data.merge(df_industry, on=['date', 'instrument'])
industry_list = df_industry['industry_level1_code'].unique()
sql = """
SELECT date, instrument, total_market_cap
FROM cn_stock_valuation
WHERE date >= '{0}' and date <= '{1}'
ORDER BY instrument, date;
""".format(start_date, end_date)
df_market_cap = dai.query(sql).df()
factor_data = factor_data.merge(df_market_cap, on=['date', 'instrument'])
factor_data['log_cap'] = np.log(factor_data.total_market_cap)
#截面中性化
def neutralize(df, col):
import warnings
warnings.filterwarnings('ignore')
import statsmodels.api as sm
col_list = [col]
for fac in col_list:
ind_dummies = pd.get_dummies(df['industry_level1_code'], prefix='industry_level1_code')
mkcap = df['log_cap']
train = pd.concat([ind_dummies,mkcap],axis=1)
X = sm.add_constant(train)
y = df[fac]
model = sm.OLS(y, X).fit()
df[fac] = model.resid
return df
res = factor_data.groupby('date').apply(neutralize, col=col)
res.sort_values(by='date', inplace=True)
res.reset_index(inplace=True, drop=True)
return res
def check_data_format(self, df):
# 检查date列是否是日期型类型
if df['date'].dtype != 'datetime64[ns]':
raise ValueError("date列的数据格式应为datetime格式")
# 检查instrument列是否是以SZ\SH结尾
if not all(df['instrument'].str.endswith('.SH') | df['instrument'].str.endswith('.SZ') | df['instrument'].str.endswith('.BJ')):
raise ValueError("instrument列的数据格式应为以.SH或.SZ或.BJ结尾的字符串")
# 检查factor列是否是浮点型数值
if df['factor'].dtype != 'float64':
raise ValueError("factor列的数据格式应为浮点型")
def stock_pool_filter(self):
pools = self.params['instruments']
if pools == "沪深300":
index_code = '000300.SH'
elif pools == "中证500":
index_code = '000905.SH'
elif pools == "中证1000":
index_code = '000852.SH'
elif pools == "全市场":
return
else:
print('请检查输入的指数池是否正确')
index_com_df = dai.query("select * from cn_stock_index_component where date >= '2015-01-01' and instrument == '%s' order by date, instrument "%index_code).df()
factor_df = self.factor_data
merge_df = pd.merge(factor_df, index_com_df, how='inner', left_on=['date','instrument'], right_on=['date', 'member_code'])[['instrument_x','date','factor']]
merge_df.rename(columns={'instrument_x':'instrument'}, inplace=True)
self.factor_data = merge_df
def get_daily_ret(self, start_date, end_date):
"""计算收益率. T0的因子对应的收益率是T+1日开盘买入,T+2开盘卖出"""
sql = f"SELECT instrument,date, (m_lead(open, 2)/ m_lead(open, 1) - 1) AS daily_ret from cn_stock_bar1d ORDER BY date, instrument;"
from datetime import datetime, timedelta
ten_days_ago_start_date = pd.Timestamp(self.start_date) - timedelta(days=10) # 往前多取10天数据
ten_days_ago_start_date = ten_days_ago_start_date.strftime('%Y-%m-%d')
price_data = dai.query(sql, filters={"date": [ten_days_ago_start_date, self.end_date]}).df()
return price_data
def get_group_data(self):
"""因子分组，因子值越大，组数越大，默认的多头组合是因子数值最大的组合"""
def cut(df, group_num=10):
"""分组"""
df = df.drop_duplicates('factor') # 删除重复值
df['group'] = pd.qcut(df['factor'], q=group_num, labels=False, duplicates='drop')
df = df.dropna(subset=['group'], how='any')
df['group'] = df['group'].apply(int).apply(str)
return df
group_data = self.merge_data.groupby('date', group_keys=False).apply(cut, group_num=self.params['group_num'])
return group_data
def get_bm_ret(self, benchmark):
if benchmark == "沪深300":
bm_code = '000300.SH'
elif benchmark == "中证500":
bm_code = '000905.SH'
elif benchmark == "中证1000":
bm_code = '000852.SH'
else:
print('请检查输入的基准代码是否正确')
# 获取基准日收益率数据
bm_sql = """
SELECT
date,instrument, (close - m_Lag(close,1)) / m_LAG(close, 1) as benchmark_ret
FROM cn_stock_index_bar1d
WHERE instrument = '%s'
AND date >= '%s' and date <='%s' ;"""%(bm_code, self.start_date, self.end_date)
bm_ret = dai.query(bm_sql).df()
return bm_ret
def get_group_cumret(self):
# 分组收益率
groupret_data = self.group_data[['date','group','daily_ret']].groupby(['date','group'], group_keys=False).apply(lambda x:np.nanmean(x)).reset_index()
groupret_data.rename(columns={0:'g_ret'}, inplace=True)
groupret_pivotdata = groupret_data.pivot(index='date', values='g_ret', columns='group')
groupret_pivotdata['ls'] = groupret_pivotdata[str(self.params['group_num']-1)] - groupret_pivotdata['0'] # 日收益率
bm_ret = self.bm_ret.set_index('date') # 基准收益率
groupret_pivotdata['bm'] = bm_ret['benchmark_ret']
groupret_pivotdata = groupret_pivotdata.shift(1) # 首日为nan，最后一日有值
self.groupret_pivotdata = groupret_pivotdata
groupcumret_pivotdata = groupret_pivotdata.cumsum() # 单利下的累积收益率
return groupcumret_pivotdata.round(4) # 数值型数据都是保留到小数点后四位
def get_Performance(self, data_type):
def get_stats(series, bm_series):
"""
series是日收益率数据, pandas.series
data_type是组合类型, 'long'、'short'、'long_short'
"""
return_ratio = series.sum() # 总收益
annual_return_ratio = series.sum() * 242 / len(series) # 年度收益
ex_return_ratio = (series-bm_series).sum() # 超额总收益
ex_annual_return_ratio = (series-bm_series).sum() * 242 / len( (series-bm_series)) # 超额年度收益
sharp_ratio = empyrical.sharpe_ratio(series, 0.035/242)
return_volatility = empyrical.annual_volatility(series)
max_drawdown = empyrical.max_drawdown(series)
information_ratio=series.mean()/series.std()
win_percent = len(series[series>0]) / len(series)
trading_days = len(series)
series = series.fillna(0)
ret_3 = series.rolling(3).sum().iloc[-1]
ret_10 = series.rolling(10).sum().iloc[-1]
ret_21 = series.rolling(21).sum().iloc[-1]
ret_63 = series.rolling(63).sum().iloc[-1]
ret_126 = series.rolling(126).sum().iloc[-1]
ret_252 = series.rolling(252).sum().iloc[-1]
return {
'return_ratio': return_ratio,
'annual_return_ratio': annual_return_ratio,
'ex_return_ratio': ex_return_ratio,
'ex_annual_return_ratio': ex_annual_return_ratio,
'sharp_ratio': sharp_ratio,
'return_volatility': return_volatility,
'information_ratio':information_ratio,
'max_drawdown': max_drawdown,
'win_percent':win_percent,
'trading_days':trading_days,
'ret_3':ret_3,
'ret_10':ret_10,
'ret_21':ret_21,
'ret_63':ret_63,
'ret_126':ret_126,
'ret_252':ret_252
}
if data_type == 'long':
perf = get_stats(self.groupret_pivotdata[str(self.params['group_num']-1)], self.groupret_pivotdata['bm'])
elif data_type =='short':
perf = get_stats(self.groupret_pivotdata['0'], self.groupret_pivotdata['bm'])
elif data_type =='long_short':
perf = get_stats(self.groupret_pivotdata['ls'], self.groupret_pivotdata['bm'])
return perf
def get_IC_data(self, data_type):
# IC
def cal_ic(df):
return df['daily_ret'].corr(df['factor'], method='spearman')
if data_type == 'all':
groupIC_data = self.group_data[['date','daily_ret','factor']].groupby('date', group_keys=False).apply(lambda x:cal_ic(x)).reset_index()
groupIC_data.rename(columns={0:'g_ic'}, inplace=True)
groupIC_data = groupIC_data.shift(1) # 首日为nan，最后一日有值
groupIC_data['ic_cumsum'] = groupIC_data['g_ic'].cumsum()
groupIC_data['ic_roll_ma'] = groupIC_data['g_ic'].rolling(22).mean()
return groupIC_data.round(4).dropna()
elif data_type == 'long':
data = self.group_data[self.group_data['group'] == str(self.params['group_num']-1)][['date','daily_ret','factor']]
groupIC_data = data.groupby('date', group_keys=False).apply(lambda x:cal_ic(x)).reset_index()
elif data_type == 'short':
data = self.group_data[self.group_data['group'] == '0'][['date','daily_ret','factor']]
groupIC_data = data.groupby('date', group_keys=False).apply(lambda x:cal_ic(x)).reset_index()
elif data_type == 'long_short':
data = self.group_data[self.group_data['group'].isin(['0',str(self.params['group_num']-1)])][['date','daily_ret','factor']]
groupIC_data = data.groupby('date', group_keys=False).apply(lambda x:cal_ic(x)).reset_index()
IC_data = groupIC_data.rename(columns={0:'g_ic'}).dropna()
ic_mean = np.nanmean(IC_data['g_ic'])
ir = np.nanmean(IC_data['g_ic']) / np.nanstd(IC_data['g_ic'])
ic_3 = IC_data['g_ic'].tail(3).mean()
ic_10 = IC_data['g_ic'].tail(10).mean()
ic_21 = IC_data['g_ic'].tail(21).mean()
ic_63 = IC_data['g_ic'].tail(63).mean()
ic_126 = IC_data['g_ic'].tail(126).mean()
ic_252 = IC_data['g_ic'].tail(252).mean()
return {
'ic':ic_mean,
'ir':ir,
'ic_3':ic_3,
'ic_10':ic_10,
'ic_21':ic_21,
'ic_63':ic_63,
'ic_126':ic_126,
'ic_252':ic_252
}
def get_Turnover_data(self, data_type):
def cal_turnover(df):
# 求每天instrument和上一日的重复元素数量
def count_repeat(s):
if s.name > 0:
return len(set(s['instrument']).intersection(set(df.loc[s.name - 1, 'instrument'])))
else:
return 0
s = df.groupby('date').apply(lambda x:x.instrument.tolist())
df = pd.DataFrame(s, columns = ['instrument']).reset_index()
# 求每天instrument有多少元素
df['instrument_count'] = df['instrument'].apply(len)
df['repeat_count'] = df.apply(count_repeat, axis=1)
df['turnover'] = 1 - df['repeat_count'] / df['instrument_count']
return np.nanmean(df['turnover'])
if data_type == 'long':
df = self.group_data[self.group_data['group'] == str(self.params['group_num']-1)]
return {'turnover':cal_turnover(df)}
elif data_type == 'short':
df = self.group_data[self.group_data['group'] == '0']
return {'turnover':cal_turnover(df)}
elif data_type == 'long_short':
long_df = self.group_data[self.group_data['group'] == str(self.params['group_num']-1)]
short_df = self.group_data[self.group_data['group'] == '0']
return {'turnover':cal_turnover(long_df) + cal_turnover(short_df)}
## 总体绩效计算
def get_whole_perf(self):
summary_df = pd.DataFrame()
for _type in ['long', 'short', 'long_short']:
dict_merged = {}
dict1 = self.get_IC_data(_type)
dict2 = self.get_Performance(_type)
dict3 = self.get_Turnover_data(_type)
dict_merged.update(dict1)
dict_merged.update(dict2)
dict_merged.update(dict3)
df = pd.DataFrame.from_dict(dict_merged, orient='index', columns=['value']).T
df['portfolio'] = _type
summary_df = summary_df.append(df)
summary_df.index = range(len(summary_df))
return summary_df.round(4)
# 按年绩效计算
def get_yearly_perf(self):
# 计算年度绩效指标
year_df = self.groupret_pivotdata.reset_index('date')
year_df['year'] = year_df['date'].apply(lambda x:x.year)
def cal_Performance(data):
series = data[str(self.params['group_num']-1)] # 只看多头组合
bm_series = data['bm']
return_ratio = series.sum() # 总收益
annual_return_ratio = series.sum() * 242 / len(series) # 年度收益
ex_return_ratio = (series-bm_series).sum() # 总收益
ex_annual_return_ratio = (series-bm_series).sum() * 242 / len(series-bm_series) # 年度收益
sharp_ratio = empyrical.sharpe_ratio(series,0.035/242)
return_volatility = empyrical.annual_volatility(series)
max_drawdown = empyrical.max_drawdown(series)
information_ratio=series.mean()/series.std()
win_percent = len(series[series>0]) / len(series)
trading_days = len(series)
perf = pd.DataFrame({
'return_ratio': [return_ratio],
'annual_return_ratio': [annual_return_ratio],
'ex_return_ratio': [ex_return_ratio],
'ex_annual_return_ratio': [ex_annual_return_ratio],
'sharp_ratio': [sharp_ratio],
'return_volatility': [return_volatility],
'max_drawdown': [max_drawdown],
'win_percent':[win_percent],
'trading_days':[int(trading_days)],
})
return perf
yearly_perf = year_df.groupby(['year'], group_keys=True).apply(cal_Performance)
yearly_perf = yearly_perf.droplevel(1).round(4) # 去掉一个level
# 计算年度IC
data = self.group_data[self.group_data['group'] == str(self.params['group_num']-1)][['date','daily_ret','factor']] # 只看多头组合
def cal_ic(df):
return df['daily_ret'].corr(df['factor'])
groupIC_data = data.groupby('date', group_keys=False).apply(lambda x:cal_ic(x)).reset_index()
IC_data = groupIC_data.rename(columns={0:'g_ic'}).dropna()
IC_data['year'] = IC_data['date'].apply(lambda x:x.year)
yearly_IC = IC_data.groupby('year').apply(lambda x:np.nanmean(x['g_ic']))
yearly_perf['ic'] = yearly_IC.round(4)
yearly_perf = yearly_perf.reset_index()
yearly_perf['year'] = yearly_perf['year'].apply(str)
return yearly_perf
def render(self):
"""图表展示因子分析结果"""
from bigcharts import opts
fields = ['portfolio','ic', 'ir', 'turnover', 'return_ratio', 'annual_return_ratio','ex_return_ratio', 'ex_annual_return_ratio', 'sharp_ratio', 'return_volatility', 'information_ratio', 'max_drawdown', 'win_percent', 'ic_252', 'ret_252']
whole_perf = self.whole_perf[fields]
c1 = bigcharts.Chart(
data=whole_perf,
type_="table",
chart_options=dict(
title_opts=opts.ComponentTitleOpts(title="整体绩效指标")
),
y=list(whole_perf.columns))
fields = ['year','ic', 'return_ratio', 'annual_return_ratio', 'ex_return_ratio', 'ex_annual_return_ratio', 'sharp_ratio', 'return_volatility',
'max_drawdown', 'win_percent', 'trading_days']
yearly_perf = self.yearly_perf[fields]
c2 = bigcharts.Chart(
data=yearly_perf,
type_="table",
chart_options=dict(
title_opts=opts.ComponentTitleOpts(title="年度绩效指标(多头组合)")
),
y=list(yearly_perf.columns))
# 绘制累积收益图
c3 = bigcharts.Chart(
data=self.group_cumret,
type_="line",
x=self.group_cumret.index,
y=self.group_cumret.columns)
_IC = np.nanmean(alpha_instance.ic['g_ic'])
_IR = np.nanmean(alpha_instance.ic['g_ic']) / np.nanstd(alpha_instance.ic['g_ic'])
abs_IC = alpha_instance.ic['g_ic'].abs()
significant_ic_ratio = abs_IC[abs_IC>=0.02].shape[0] / abs_IC.shape[0]
c4 = bigcharts.Chart(
data=pd.DataFrame({'IC':[np.round(_IC,4)], '|IC|>0.02':[np.round(significant_ic_ratio,4)], 'IR':[np.round(_IR,4)]}),
type_="table",
chart_options=dict(
title_opts=opts.ComponentTitleOpts(title="IC分析指标")
),
y=['IC','|IC|>0.02','IR'],
)
# 绘制每期IC时序图
c5 = bigcharts.Chart(
data=self.ic,
type_="bar",
x='date',
y=['g_ic', 'ic_roll_ma'],
chart_options=dict(
title_opts=opts.TitleOpts(
title="IC曲线",
subtitle="每日IC、累计IC、近22日IC均值",
pos_left="center",
pos_top="top",
),
legend_opts=opts.LegendOpts(
is_show=False, # 不显示图例
),
extend_yaxis=[opts.AxisOpts()]
)
)
# 绘制IC累计曲线图
c6 = bigcharts.Chart(
data=self.ic,
type_="line",
x='date',
y=['ic_cumsum'],
chart_options=dict(
title_opts=opts.TitleOpts(
title="IC累积曲线",
pos_left="center",
pos_top="top",
),
legend_opts=opts.LegendOpts(
is_show=False, # 不显示图例
)
),
series_options={"ic_cumsum": {"yaxis_index": 1}}
)
c5_6 = bigcharts.Chart(data = [c5, c6], type_ = "overlap",)
top_factor_df = self.factor_data[self.factor_data['date'] == self.end_date].round(4) # 最后一天因子数据
top_factor_df['date'] = top_factor_df['date'].apply(lambda x:x.strftime('%Y-%m-%d'))
# 按照 factor 列升序排序，获取最小的10行数据
df_sorted_min = top_factor_df.sort_values('factor').head(self.top_n_ins)
# 按照 factor 列降序排序，获取最大的10行数据
df_sorted_max = top_factor_df.sort_values('factor', ascending=False).head(self.top_n_ins)
c7 = bigcharts.Chart(
data=df_sorted_max,
type_="table",
chart_options=dict(
title_opts=opts.ComponentTitleOpts(title="因子值最大的%s只标的"%self.top_n_ins)
),
y=['date','instrument','factor'],
)
c8 = bigcharts.Chart(
data=df_sorted_min[['date','instrument','factor']],
type_="table",
chart_options=dict(
title_opts=opts.ComponentTitleOpts(title="因子值最小的%s只标的"%self.top_n_ins)
),
y=['date','instrument','factor'],
)
c_set = bigcharts.Chart([c1, c2, c3, c4, c5_6, c7, c8], type_="page").render(display=False)
from IPython.display import display
display(c_set)
return c_set.data
Step 3 因子分析
alpha_instance = AlphaMiner(params=params, factor_data=factor_data)
report_html = alpha_instance.render()
Step 4 因子提交
perf_df = alpha_instance.whole_perf
perf_dict = {}
perf_dict['IC'] = np.nanmean(alpha_instance.ic['g_ic'])
for i in perf_df.index:
df = perf_df.iloc[i]
flag = df['portfolio']
for c in perf_df.columns:
perf_dict['%s_%s'%(flag, c)] = df[c]
perf_dict.update(alpha_instance.params)
for key, value in perf_dict.items():
if isinstance(value, (int, float)):
if np.isinf(value) or np.isnan(value):
perf_dict[key] = None
from bigalpha import factors
# 提交到公共空间
factors.submit_factor(
id = 'alpha_6100', #因子分析ID
performance_index = perf_dict, # 因子绩效
performance_report = report_html, # 因子详情页面
metadata = {}, # 可以先不填
docs = {'因子类型':'Alpha101因子'},
name = "Alpha_6100", # 中文名
desc = """Alpha_6100的定义是：取收盘价与最低价差值减去最高价与收盘价差值除以最高价与最低价差值乘以成交量，对所得值作两次行业中性化（行业分类参考申万2021一级行业分类），再缩放所得值并乘以1.5；另取收盘价与20日日均成交金额的五日相关系数减去30日收盘价最小值所在天数的横截面排名，对所得值作行业中性化处理，再作缩放处理；前值减去后值再乘以成交量除以20日日均成交金额乘以1，最后用0减去上述所得值得到因子值。（日期小数四舍五入处理）（Alpha#100: (0 - (1 * (((1.5 * scale(indneutralize(indneutralize(rank(((((close - low) - (high -
close)) / (high - low)) * volume)), IndClass.subindustry), IndClass.subindustry))) -
scale(indneutralize((correlation(close, rank(adv20), 5) - rank(ts_argmin(close, 30))),
IndClass.subindustry))) * (volume / adv20))))）
""",
# datasource_id = 'cn_stock_alphas_base', # 表名
# datasource_column = 'return20', # 字段名
factor_sql = sql,
)
print('因子入库完成!')
\