择时策略 —— 基于北上资金的沪深300指数择时

1、策略概述

北向资金是指通过港交所流入A 股的资金，资金来源可能是外资、港资，也有可能是国内借道香港的“出口转内销”资金，托管方是在香港营业的银行或者券商。北向资金总体上主要流向白马股，它们的规模虽然在A股市场中的比例不大，但是从2014年底以来，长期持续流入的北向资金在一定程度上影响了A股市场的因子风格，而北向资金自己也在其中获取了丰厚的回报。
因此，参考北向资金进行投资，是一个值得探讨的方向。有的研究报告将北向资金又分成托管在银行的资金和托管在券商的资金，认为前者持仓相对较长，是白马股的坚定持有者，后者更类似短期炒作资金，虽然交易标的物也是白马股，但是并不是被动持仓，而会根据市场的短期情况高抛低吸。由于数据源的原因，无法拿到北向资金的托管数据，因此本贴子只关注总体情况。

2、策略规则

1、取北向资金的每日总数，生成其 bollinger 指标，公式为：

mid = sma(north_money, 252)
up_band = mid + std(north_money, 252) * 1.5
dn_band = mid - std(north_money, 252) * 1.5

2、当某日的北上资金总额上穿 up_band，第二天开盘时在沪深300指数上开多；下穿 dn_band 时，第二天开盘时在沪深300指数上平多。
3、暂不考虑手续费和滑点

3、策略实现

1、数据从本地数据库中获取。
2、策略在自建的事件驱动框架上实现，相关介绍见自建基于事件驱动的策略回测、因子/指标计算框架简介
3、show me the code

from abc import ABC
from os import getcwd
from os.path import basename, abspath, dirname
from pandas import Series
from time import strftime
from typing import List, Dict
from numpy import nanfrom core.const import Product, Slippage, db_dict
from core.utility import Timer, ColorLogger
from strategy.strategy_equity import StrategyEquity
from data_center.access_data_sqlite import connect_ts, connect_db, connect_dolphindb_dailyclass NorthMoneyTiming(StrategyEquity, ABC):def __init__(self,clog: any,start: str, end: str, account: List, geteway: str = 'quandomo'):super(NorthMoneyTiming, self).__init__(clog, start, end, account, geteway)self.set_trading_cost('zero')# 指标值self.indicator_value = self.get_indicator()# 取指数数据cur_sql = f"select ts_code,trade_date,open,high,low,close,vol from index_daily " \f"where ts_code='000300.SH'"self.index_daily = self.db_ts.get_data(cur_sql, 'frame', 'trade_date')# self.index_daily = self.index_daily.reset_index().set_index(['ts_code', 'trade_date'])self.index_daily = self.index_daily.reset_index()self.index_daily['sma20'] = self.index_daily['close'].rolling(20).mean()def get_indicator(self) -> Series:"""取指标值"""cur_sql = f"select trade_date,north_money from moneyflow_hsgt " \f"where trade_date between '{self.start}' and '{self.end}' and north_money != 0"cur_indicator = self.origin_db_list[0].get_data_from_sqlite(cur_sql, 'frame').set_index('trade_date')cur_indicator['ma'] = cur_indicator['north_money'].rolling(252).mean()cur_indicator['std'] = cur_indicator['north_money'].rolling(252).std()cur_indicator['up_band'] = cur_indicator['ma'] + cur_indicator['ma'] * 1.5cur_indicator['dn_band'] = cur_indicator['ma'] - cur_indicator['ma'] * 1.5return cur_indicatordef get_trade_signal(self, dt: str) -> int:""" 生成交易信号 """signal = 0try:if self.pre_dt and self.indicator_value.loc[self.pre_dt, 'ma'] is not nan:if self.indicator_value.loc[self.pre_dt, 'north_money'] < self.indicator_value.loc[self.pre_dt, 'up_band'] \and self.indicator_value.loc[dt, 'north_money'] >= self.indicator_value.loc[dt, 'up_band']:signal = 1elif self.indicator_value.loc[self.pre_dt, 'north_money'] >= self.indicator_value.loc[self.pre_dt, 'dn_band'] \and self.indicator_value.loc[dt, 'north_money'] < self.indicator_value.loc[dt, 'dn_band']:signal = -1except (KeyError, ValueError, IndexError, NameError):passreturn signaldef handle_bar(self, event_bar):""" 逐个 bar 运行回测 """self.activate_trade_signal = Falseself.trade_signal[event_bar.dt] = self.get_trade_signal(event_bar.dt)if self.trade_signal[event_bar.dt] > 0 and self.market_position('000300.SH') < 1:self.position_lst[event_bar.dt] = ['000300.SH']self.activate_trade_signal = True   # 已有交易，本日不再做资产调整elif self.trade_signal[event_bar.dt] < 0 and self.market_position('000300.SH') > 0:self.position_lst[event_bar.dt] = []self.activate_trade_signal = True  # 已有交易，本日不再做资产调整def handle_trade(self, event_trade):""" 委托成交 """super(NorthMoneyTiming, self).handle_trade(event_trade)self.context.logger.info(f'-- [on_trade] {event_trade.data.symbol} 在 {event_trade.data.datetime} 的委托单 'f'{event_trade.data.order_id} [{event_trade.data.comments}]，开盘后成交，'f'成交价格是 {event_trade.data.price}, 成交量是 {round(event_trade.data.volume, 2)}')if __name__ == "__main__":root_path = abspath(dirname(getcwd()))      # 取本脚本文件所在目录的父目录export_path = '/export/'# 新建 log 文件log_path = '/log/'log_name = f"{basename(__file__)[:-3]} {strftime('%Y-%m-%d_%H%M%S')}.log"color_logger = ColorLogger(log_name=root_path + log_path + log_name)strat_account = [{'name': 'MyAccount', 'equity': 20000000}]benchmark = '000300.SH'cur_start_date = '20141117'cur_end_date = '20210407'   # '20210223'# end_date = f"{strftime('%Y%m%d')}"fields = ['open', 'high', 'low', 'close', 'volume']strategy_type = 'timing_only'# 显示运行完整个策略所需时间time_test = Timer(True)with time_test:color_logger.info(f'**************** 开始运行策略 *****************')color_logger.info(f'')# 策略实例化，并给变量赋值trial_strategy = NorthMoneyTiming(clog=color_logger,start=cur_start_date,end=cur_end_date,account=strat_account)trial_strategy.run_strategy(fields, strategy_type)trial_strategy.performance_analysis(root_path + export_path, benchmark)  # 绩效分析trial_strategy.plot_results(root_path + export_path, basename(__file__)[:-3])

4、策略绩效

1、策略净值曲线

2、绩效统计

3、策略的逐月回报统计

参考资料：
1、2020-08-03，北向资金走向预示市场短期或震荡——华泰金工量化资产配置 7月月报