[Python知识库]python量化交易第五章

一.什么是动量策略

1.1概念：
预先对股票收益和交易量设定过滤准则，当股票收益或股票收益和交易量同时满足过滤准则就买入（做多）或卖出（做空）股票的投资策略。
核心：以股票的历史收益率作为主要的交易原则
1.2动量策略的设计思路

正向策略：涨的还会涨，跌的还会跌，买入涨最多的，卖出跌最多的，利用市场对信息的反应不足。
反向策略:涨太多了会跌，跌太多了会涨，买入跌最多的，卖出涨最多的，市场对信息反应过度
1.3动量策略的实现步骤

1. 确定交易对象：股票池，考虑流动性（沪深300，创业板）
2. 选定业绩评价周期：过去1-12个月
3. 计算形成期收益率：过去N个月的收益率
4. 对收益率进行排序：最佳—赢家组合，最差~输家组合
5. 确定持仓/换仓周期：1个月，可自定义测算
6. 连续或间隔一段时间，不断重复2-5行为
7. 计算动量/反向策略各持有期的回报率
8. 计算t/p统计值，判断是否存在动量效应

二.动量策略：筛选股票池

• 流动性：成交活跃，买入卖出顺畅
• 基本面：行业，营收，盈利增速，现金流，负债
• 标的价格: 1手起买起卖

import data.stock as st

def get_data(index_symbol='000300.XSHG'):
    # 获取股票列表代码：沪深300持有个股，创业板，上证
    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,拼接后的数据
    """
    # 获取股票列表代码：沪深300持有个股，创业板，上证
    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())
        #拼接多个股票的收盘价：日期 股票A收盘价 股票B收盘价...
        data.colimns=[code]
        pd.concat([data_concat,data],axis=1)
    # 预览股票数据
    # print(data_concat.tail())
    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()
    #计算过去N个月的收益率=期末值/期初值-1=（期末-期初）/期初
    #optional:对数收益率=log(期末值/期初值)
    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,拼接后的数据
    """
    # 获取股票列表代码：沪深300持有个股，创业板，上证
    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())
        #拼接多个股票的收盘价：日期 股票A收盘价 股票B收盘价...
        data.columns=[code]
        pd.concat([data_concat,data],axis=1)
    # 预览股票数据
    # print(data_concat.tail())
    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()
    #计算过去N个月的收益率=期末值/期初值-1=（期末-期初）/期初
    #optional:对数收益率=log(期末值/期初值)
    shift_return=data_month/data_month.shift(shift_n)-1
    print(shift_return)
    #生成交易信号：收益率排前n的>赢家组合>买入1，排最后n个>输家>卖出-1
    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(data_month.head())
    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)
    #对data的每一行进行遍历，找出里面的最大值，并利用bool函数标注0或1信号
    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)