策略名称：北向资金择时策略
策略说明：
本策略基于北向资金（沪股通和深股通）的净流入数据，通过计算短期和长期的指数加权移动平均线（EWMA）以及标准差，构建了一个择时信号。策略的核心思想是，当北向资金的短期净流入显著高于长期净流入时，认为市场可能处于上升趋势，选择开仓；反之，当短期净流入显著低于长期净流入时，认为市场可能处于下降趋势，选择平仓。
策略逻辑：
1. **数据获取**：
- 通过`query_northmoney`函数获取北向资金的买入金额、卖出金额等数据。
- 计算每日的北向资金净流入（买入金额 - 卖出金额）。
2. **信号生成**：
- 计算北向资金净流入的长期（71天）和短期（6天）指数加权移动平均线（EWMA）。
- 计算北向资金净流入的标准差（基于长期窗口）。
- 生成择时信号：`(短期EWMA - 长期EWMA) / 标准差`。
- 当信号值大于设定的高阈值（-0.499）时，认为市场处于上升趋势，选择开仓；当信号值小于设定的低阈值（-0.159）时，认为市场处于下降趋势，选择平仓。
3. **交易执行**：
- 在每日的`handle_data`函数中，根据生成的择时信号决定是否开仓或平仓。
- 开仓时，买入标的ETF（沪深300ETF，代码：510300.XSHG）；平仓时，卖出所有持仓。
策略参数：
- **标的ETF**：沪深300ETF（510300.XSHG）
- **高阈值（threshold_h）**：-0.499
- **低阈值（threshold_l）**：-0.159
- **长期窗口（long_period）**：71天
- **短期窗口（short_period）**：6天
策略特点：
- **基于北向资金**：北向资金通常被视为“聪明钱”，其流入流出情况对市场有一定的预示作用。
- **择时信号**：通过短期和长期净流入的对比，捕捉市场的趋势变化。
- **动态调整**：根据市场情况动态调整仓位，避免在市场下跌时持有过多仓位。
适用场景：
- **市场趋势明显时**：当市场处于明显的上升或下降趋势时，策略表现较好。
- **北向资金活跃时**：在北向资金流入流出较为活跃的市场环境下，策略的信号更为有效。
风险提示：
- **滞后性**：择时信号基于历史数据生成，可能存在一定的滞后性。
- **市场波动**：在市场波动较大时，策略可能会频繁开仓平仓，增加交易成本。
- **北向资金数据限制**：北向资金数据从2014年11月开始，策略在早期数据不足时可能表现不稳定。
总结：
本策略通过北向资金的净流入数据构建择时信号，旨在捕捉市场的趋势变化，适合在趋势明显的市场环境下使用。策略的核心在于通过短期和长期净流入的对比，动态调整仓位，以期在市场上升时获得收益，在市场下跌时减少损失。
策略代码：
# 标题：【复现】北向资金交易能力一定强吗
# 作者：Hugo2046
'''
Author: Hugo
Date: 2020-09-28 21:52:43
LastEditTime: 2020-09-28 22:10:35
LastEditors: Hugo
Description: 根据最优参数进行择时回测,注意北流数据2014年11月才有数据
关键函数解释：
distributed_query 用于若干数据查询限制
query_northmoney 用于查询北向资金数据
get_net_north_flow 用于计算北向资金净流量
get_north_factor 是否开仓 大于阈值开仓 小于阈值平仓
'''
from jqdata import *
import numpy as np
import pandas as pd
enable_profile() # 开启性能分析
def initialize(context):
set_params()
set_variables()
set_backtest()
def set_params():
g.target_etf = '510300.XSHG' # 标的为HS300
# 设置参数
g.params = {'threshold_h': -0.4991838749623957,
'threshold_l': -0.15982912319375642,
'long_period': 71,
'short_period': 6}
def set_variables():
pass
def set_backtest():
set_option("avoid_future_data", True) # 避免数据
set_option("use_real_price", True) # 真实价格交易
set_benchmark('000300.XSHG') # 设置基准
#log.set_level("order", "debuge")
log.set_level('order', 'error')
# 每日盘前运行
def before_trading_start(context):
set_slip_fee(context)
# 设置不同时期手续费
def set_slip_fee(context):
# 将滑点设置为0
set_slippage(FixedSlippage(0))
# 根据不同的时间段设置手续费
dt = context.current_dt
if dt > datetime.datetime(2013, 1, 1):
set_commission(PerTrade(buy_cost=0.0003, sell_cost=0.0013, min_cost=5))
elif dt > datetime.datetime(2011, 1, 1):
set_commission(PerTrade(buy_cost=0.001, sell_cost=0.002, min_cost=5))
elif dt > datetime.datetime(2009, 1, 1):
set_commission(PerTrade(buy_cost=0.002, sell_cost=0.003, min_cost=5))
else:
set_commission(PerTrade(buy_cost=0.003, sell_cost=0.004, min_cost=5))
########################################### 数据获取 ####################################################
def distributed_query(query_func_name, start: str, end: str, limit=3000, **kwargs) -> pd.DataFrame:
'''用于绕过最大条数限制'''
days = get_trade_days(start, end)
n_days = len(days)
if len(days) > limit:
n = n_days // limit
df_list = []
i = 0
pos1, pos2 = n * i, n * (i + 1) - 1
while pos2 < n_days:
df = query_func_name(
start=days[pos1],
end=days[pos2],
**kwargs)
df_list.append(df)
i += 1
pos1, pos2 = n * i, n * (i + 1) - 1
if pos1 < n_days:
df = query_func_name(
start=days[pos1],
end=days[-1],
**kwargs)
df_list.append(df)
df = pd.concat(df_list, axis=0)
else:
df = query_func_name(
start=start, end=end, **kwargs)
return df
def query_northmoney(start: str, end: str, fields: list) -> pd.DataFrame:
'''北向资金成交查询'''
select_type = ['沪股通', '深股通']
select_fields = ','.join([f"finance.STK_ML_QUOTA.%s" % i for i in fields])
df_list = []
for types in select_type:
q = query(select_fields).filter(finance.STK_ML_QUOTA.day >= start,
finance.STK_ML_QUOTA.day <= end,
finance.STK_ML_QUOTA.link_name == types)
df_list.append(finance.run_query(q))
return pd.concat(df_list)
# 构造北流净值
def get_net_north_flow(watch_date: str, N: int) -> pd.Series:
begin = get_trade_days(end_date=watch_date, count=N)[
0].strftime('%Y-%m-%d')
# 获取北流相关数据
fields = ['day', 'link_name', 'buy_amount', 'sell_amount', 'sum_amount']
north_money = distributed_query(
query_northmoney, begin, watch_date, fields=fields)
# 日度合计各项指标
daily_northmoney = north_money.groupby('day').sum()
daily_northmoney.index = pd.to_datetime(daily_northmoney.index)
# 计算净流
return daily_northmoney['buy_amount'] - daily_northmoney['sell_amount']
# 信号构造
def get_north_factor(north_flow: pd.Series, params: dict) -> bool:
l = north_flow.ewm(span=params['long_period'], adjust=False).mean()
s = north_flow.ewm(span=params['short_period'], adjust=False).mean()
north_factor = (s - l) / north_flow.rolling(params['long_period']).std()
record(north_factor=north_factor.iloc[-1],
h=params['threshold_h'], l=params['threshold_l'])
if north_factor.iloc[-1] > params['threshold_h']:
log.info('需要开仓,信号:%.4f,阈值:%.4f' %
(north_factor.iloc[-1], params['threshold_h']))
return True
elif north_factor.iloc[-1] < params['threshold_l']:
log.info('需要平仓,信号:%.4f,阈值:%.4f' %
(north_factor.iloc[-1], params['threshold_h']))
return False
else:
log.info('不做操作,信号:%.4f,阈值:%.4f' %
(north_factor.iloc[-1], params['threshold_h']))
return True
########################################### 交易 ####################################################
def handle_data(context, data):
bar_time = context.previous_date.strftime('%Y-%m-%d')
north_flow = get_net_north_flow(bar_time, 300)
is_trade = get_north_factor(north_flow, g.params)
if is_trade and len(context.portfolio.long_positions) == 0:
log.info('执行开仓操作')
order_target_value(g.target_etf, context.portfolio.total_value)
if not is_trade and len(context.portfolio.long_positions) != 0:
log.info('执行平仓操作')
order_target(g.target_etf, 0)
2025-02-20 