技术信号选股

写单个选股器

# encoding:utf-8
from datetime import datetime, timedelta
import pandas as pd
from fxdayu_data import DataAPI
from talib.abstract import MA
from fxdayu_alphaman.selector.selector import Selector
class DayMA(Selector):
    # 在此处设置选股器相关参数
    fast = 5
    slow = 10
    max_window = 30 #最大回溯单位时间数 可选 默认100
    def calculate_MA_signal(self, data):
        candle_data = data[1].dropna()
        if len(candle_data) == 0:
            return
        fast = MA(candle_data, timeperiod=self.fast)
        slow = MA(candle_data, timeperiod=self.slow)
        f_over_s = fast - slow >0
        f_under_s = fast - slow <=0
        cross_over = (f_under_s.shift(1)) * f_over_s
        cross_under = - (f_over_s.shift(1))*f_under_s
        choice = pd.DataFrame(cross_over + cross_under)
        choice.columns = [data[0],]
        return choice
    def execute(self, pool, start, end, data=None, data_config=None):
        """
        计算选股结果
        :param pool: list 待选股票池
               data：进行选股操作所需要的数据
               start:datetime 选股索引范围起始时间
               end：datetime 选股索引范围结束时间
        :return:selector_result:A MultiIndex Series indexed by date (level 0) and asset (level 1), containing
        the values mean whether choose the asset or not.
            -----------------------------------
                date    |    asset   |
            -----------------------------------
                        |   AAPL     |   1
                        -----------------------
                        |   BA       |  1
                        -----------------------
            2014-01-01  |   CMG      |   1
                        -----------------------
                        |   DAL      |  0
                        -----------------------
                        |   LULU     |   -1
                        -----------------------
        """
        if not data:
            data = DataAPI.get(symbols=tuple(pool),
                               start=start-timedelta(days=self.max_window) ,
                               end = end,
                               **data_config)
        selector_result = map(self.calculate_MA_signal, data.iteritems())
        selector_result = pd.concat(selector_result, axis=1).stack()
        selector_result.index.names = ["date","asset"]
        return selector_result.loc[start:end]

选股器单元测试

def unit_test():
    #选股器单元测试——通过Admin
    selector_admin = Admin()
    result = selector_admin.instantiate_selector_and_get_selector_result("DayMA",codes,start,end,data_config=data_config)
#     print(selector_admin)
    print(result)
    return(selector_admin , result)
####### test
selector_admin , result = unit_test()

选股绩效测试

def test_performance(selector_admin,result):
    performance = selector_admin.calculate_performance("DayMA",
                                                       result[result>0], #结果大于0的（选出的）
                                                       start,
                                                       end,
                                                       periods=periods,
                                                       benchmark_return=hs300_return)
    print(performance["mean_return"]) #选股策略平均持有收益
    print(performance["key_performance_indicator"]) #关键性绩效指标
    print(performance["upside_return"]) #上行收益
    print(performance["upside_distribution_features"]) #上行收益特征
    print(performance["downside_return"]) #下行收益
    print(performance["downside_distribution_features"]) #下行收益特征
    # 支持类属性的调用方式
    print(performance.holding_return) #持有收益
    print(performance.holding_distribution_features) #持有收益特征
    return performance
performance = test_performance(selector_admin,result)

优化选股器参数

para_range_dict = {"fast":range(2,5,1),"slow":range(10,15,1)}
results_list, para_dict_list = selector_admin.enumerate_parameter("DayMA",
                                                                  para_range_dict,
                                                                  codes,
                                                                  start,
                                                                  end,
                                                                  data_config=data_config,
                                                                  parallel=False)
# 批量计算多个不同参数下方案的绩效表现
strategy_name_list = []
strategy_result_list = []
for i in range(len(results_list)):
    print("\n")
    strategy_name = "DayMA+" + str(para_dict_list[i])
    print(strategy_name)
    strategy_name_list.append(strategy_name)
    strategy_result = results_list[i][results_list[i]>0]
    print(strategy_result)
    strategy_result_list.append(strategy_result) #只记录大于0的结果
performance_list = selector_admin.show_strategies_performance(strategy_name_list,
                                                              strategy_result_list,
                                                              start,
                                                              end,
                                                              periods=periods,
                                                              benchmark_return=hs300_return,
                                                              parallel=False)
for perf in performance_list:
    print("\n")
    print(perf.strategy_name)
    print(perf.key_performance_indicator)