一.什么是动量策略
1.1概念：
预先对股票收益和交易量设定过滤准则，当股票收益或股票收益和交易量同时满足过滤准则就买入（做多）或卖出（做空）股票的投资策略。
核心：以股票的历史收益率作为主要的交易原则
1.2动量策略的设计思路
正向策略：涨的还会涨，跌的还会跌，买入涨最多的，卖出跌最多的，利用市场对信息的反应不足。
反向策略:涨太多了会跌，跌太多了会涨，买入跌最多的，卖出涨最多的，市场对信息反应过度
1.3动量策略的实现步骤
- 确定交易对象：股票池，考虑流动性（沪深300，创业板）
- 选定业绩评价周期：过去1-12个月
- 计算形成期收益率：过去N个月的收益率
- 对收益率进行排序：最佳—赢家组合，最差~输家组合
- 确定持仓/换仓周期：1个月，可自定义测算
- 连续或间隔一段时间，不断重复2-5行为
- 计算动量/反向策略各持有期的回报率
- 计算t/p统计值，判断是否存在动量效应
二.动量策略：筛选股票池
- 流动性：成交活跃，买入卖出顺畅
- 基本面：行业，营收，盈利增速，现金流，负债
- 标的价格: 1手起买起卖
import data.stock as st
def get_data(index_symbol='000300.XSHG'):
stocks=st.get_index_list(index_symbol)
for code in stocks:
data=st.get_csv_price(code,'2019-01-01','2021-01-04')
print("================",code)
print(data.tail())
def momentum():
return 0
if __name__=="__main__":
get_data()
def get_index_list(index_symbol='000300.XSHG'):
"""
获取指数成分股，指数代码查询
:param index_symbol:https://www.joinquant.com/indexdata
:return:list，成分股代码
"""
stocks=get_index_stocks(index_symbol)
return stocks
三.动量策略：计算动量因子
import pandas as pd
import data.stock as st
def get_data(start_date,end_date,use_cols,index_symbol='000300.XSHG'):
"""
获取股票收盘价数据，并拼接为一个df
:param start_date:
:param end_date:
:param use_cols:
:param index_symbol:
:return datac_concat:df,拼接后的数据
"""
stocks=st.get_index_list(index_symbol)
data_concat=pd.DataFrame()
for code in stocks[0:10]:
data=st.get_csv_price(start_date,end_date,use_cols,code)
print("================",code)
print(data.tail())
data.colimns=[code]
pd.concat([data_concat,data],axis=1)
return data_concat
def momentum(data_concat,shift_n=1):
"""
:param data_concat:
:param shift_n:表示业绩统计周期(单位：月)
:return:
"""
data_concat.index=pd.to_datetime(data_concat.index)
data_month=data.concat.resample('M').last()
print(data_month.head())
shift_return=data_month/data_month.shift(shift_n)-1
print(data_month.head())
return shift_return
if __name__=="__main__":
data=get_data('2019-01-01','2021-01-04',['date','close'])
momentum(data,shift_n=1)
三.动量策略：生成交易信号
import pandas as pd
import numpy as np
import data.stock as st
def get_data(start_date,end_date,use_cols,index_symbol='000300.XSHG'):
"""
获取股票收盘价数据，并拼接为一个df
:param start_date:
:param end_date:
:param use_cols:
:param index_symbol:
:return datac_concat:df,拼接后的数据
"""
stocks=st.get_index_list(index_symbol)
data_concat=pd.DataFrame()
for code in stocks[0:10]:
data=st.get_csv_price(start_date,end_date,use_cols,code)
print("================",code)
print(data.tail())
data.columns=[code]
pd.concat([data_concat,data],axis=1)
return data_concat
def momentum(data_concat,shift_n=1,top_n=2):
"""
:param data_concat:
:param shift_n:表示业绩统计周期(单位：月)
:return:
"""
data_concat.index=pd.to_datetime(data_concat.index)
data_month=data.concat.resample('M').last()
shift_return=data_month/data_month.shift(shift_n)-1
print(shift_return)
buy_signal=get_top_stocks(shift_return,top_n)
sell_signal=get_top_stocks(-1*shift_return,top_n)
signal=buy_signal-sell_signal
print(signal)
exit()
print(shift_return.head())
return shift_return
def get_top_stocks(data,top_n):
"""
找到前n位的极值，并转换为信号返回
:param data:df
:param top_n:int，表示要产生信号的个数
:return signals:df,返回0-1信号数据表
"""
signals=pd.DataFrame(index=data.index,columns=data.columns)
for index,row in data.iterrows():
print(row.isin(row.nlargest(top_n)).astype(np.int))
return signals
if __name__=="__main__":
data=get_data('2019-01-01','2021-01-04',['date','close'])
momentum(data,shift_n=1)
|