【重磅更新】回测代码直接实盘交易 只需一分钟!

用户头像神盾局量子研究部
2023-05-15 发布

背景与目的

之前我们有了策略回测代码,到实盘要经过熟悉实盘API、写代码、调试代码的环节,大概还需要1-2周的时间才能实盘,有非常多的用户到这一步 会束手无策,甚至放弃!

现在,有了回测代码直接实盘的功能,可以省去这个步骤,让刚入门的朋友也可以直接拿回测代码进行实盘了。很棒!为我们的工程师点赞!

此外还新增了一些接口,方便实盘交易。

不断降低实盘的门槛是我们的目标,如果您有任何好的想法意见请随时留言!

本功能需要重启研究环境才能生效!

策略实盘交易(回测代码1分钟实盘)

  • ?调用方法:
    research_trade(
        name,
        source_code,
        capital_base=100000,
        frequency='DAILY',
        stock_market='STOCK',
        benchmark=None,
        trade_api=None,
        signal_mode=True,
        dry_run=False,
        recover_dt=False,
    )
    
    
  • ?参数说明:
    • name:str,策略名称,会在./persist/下生成一个同名目录,用于存放持久化的策略信息
    • source_code:str,策略代码,可从策略研究模块中直接复制,代码置于"""..."""中
    • capital_base: float,初始资金量
      • 如果接入了TradeAPI对象,且 signal_mode=False,那么此参数无意义
    • frequency: str,策略频率,'DAILY'或'MINUTE'
    • stock_market: str,策略类型,默认'STOCK'
    • benchmark: str,基准指数
    • trade_api: TradeAPI对象,绑定需要仿真交易的资金账号
      • 如果不传入TradeAPI对象,即 trade_api=None,此时为模拟交易
      • 如果传入TradeAPI对象,此时为仿真交易
    • signal_mode: bool,(新增)默认为 True
      • signal_mode=True,此时策略实际上运行的时初始资金为capital_base的模拟交易,context、get_orders等方法返回的结果均为模拟交易中计算的数据,与资金账号的数据无关;策略下单在模拟交易撮合成交后,才会通过trade_api下单至柜台
      • signal_mode=False,此时策略中context、get_orders等方法返回的结果均为从 柜台查询,策略下单也会直接下至柜台
    • dry_run: bool,试运行,立即返回,默认为 False
    • recover_dt: bool或 str,(新增)是否断点运行,默认为 False
      • recover_dt=False,从当前时点开始执行,不从断定运行
      • recover_dt=True,从上次策略结束时点开始运行
      • recover_dt='today',从当日开始运行,此模式下只会补执行 before_tradingopen_auctionhandle_bar依旧从当前时间开始执行
      • recover_dt='yyyyMMdd HH:mm',从指定时间开始运行
  • ?️ 返回值:
  • ?作用:
    • 模拟交易:撮合机制与回测相同
    • 仿真交易:通过仿真柜台撮合,更贴近真实交易环境
  • ❗注意事项:
    • 策略需在9:00前开启运行,否则在未设置recover_dt的情况下,会跳过before_trading等步骤
    • 初始化TradeAPI时需要指定下单策略order_policy,MarketPolicy为市价下单;LimitPolicy为限价下单。如未指定,由于策略下单时使用均价,可能存在多位小数,最终实盘账户下单的时候可能产生废单
    • signal_mode=True时,如想在context中获得仿真账号的持仓、资金等数据,可以使用同步函数 sync_trade_api()
  • ?示例:
    from tick_trade_api import TradeAPI
    #初始化TradeAPI时需要指定下单策略,MarketPolicy为市价下单;LimitPolicy为限价下单
    trade_api=TradeAPI('69271711',order_policy=MarketPolicy)
    
    source_code="""
    # 股票策略模版
    def init(context):
        pass
    
    # 盘前执行
    def before_trading(context):
        pass
    
    # 开盘时运行函数
    def handle_bar(context, bar_dict):
        order_id = order('000001.SZ', 100)
        print(get_orders())
        try:
            cancel_order(order_id)
        except:
            print('撤单失败')
        print(get_open_orders())
        print(get_tradelogs())
        print(context.portfolio.stock_account)
        print(context.portfolio.positions)
    """
    rtrade = research_trade(
        '研究环境策略',
         source_code,
         frequency='MINUTE', 
         trade_api=trade_api,
         signal_mode=False,
         recover_dt='today'
    )
    
    

trade_api=TradeAPI('69271711',order_policy=MarketPolicy) 中的账号是模拟资金账号或者是实盘资金账号。

df54bf7a47071c5e416a1209b16febaa.png

其他更新

这次还增加了几个功能

策略框架中增加 :

  • cancel_order_all() 全撤
  • get_tradelogs()获取当日全部成交订单
  • get_orders() 获取委托,和get_order()一致,主要时和tradeapi中函数名对齐

tradeapi增加:

  • get_open_orders() 获取当日未成订单
  • cancel_order_all() 全撤

评论

用户头像
2023-12-09 16:30:23

双均线+ATR交易系统(适配2025年回测时间)

def initialize(context):

系统参数配置

context.N1 = 10 # 短期均线周期
context.N2 = 20 # 长期均线周期
context.ATR_N = 14 # ATR计算周期
context.ATR_SL = 2 # 止损倍数(ATR)
context.ATR_TP = 6 # 止盈倍数(ATR)
context.risk_ratio = 0.01 # 单笔风险比例(1%)

【关键】订阅2025年的有效主力合约

context.symbols = ['SC2503', 'CU2503', 'J2503', 'P2503']
for sym in context.symbols:
subscribe(sym)

ATR指标计算函数

def ATR(df, n):
df['tr'] = df.apply(lambda row: max(row['high'] - row['low'],
abs(row['high'] - row['close'].shift(1)),
abs(row['low'] - row['close'].shift(1))), axis=1)
df['atr'] = df['tr'].rolling(n).mean()
return df['atr'].iloc[-1]

核心交易逻辑

def handle_data(context, data):
for symbol in context.symbols:

获取日线数据(趋势判定)

df_daily = get_price(symbol, end_date=context.current_dt,
frequency='daily', count=50, fields=['close', 'high', 'low'])
if len(df_daily) < 20:
continue

计算日线均线与ATR

df_daily['MA10'] = df_daily['close'].rolling(context.N1).mean()
df_daily['MA20'] = df_daily['close'].rolling(context.N2).mean()
atr_daily = ATR(df_daily, context.ATR_N)

判定日线趋势

ma10_up = (df_daily['MA10'].iloc[-1] > df_daily['MA10'].iloc[-2] > df_daily['MA10'].iloc[-3])
ma20_up = (df_daily['MA20'].iloc[-1] > df_daily['MA20'].iloc[-2] > df_daily['MA20'].iloc[-3])
ma10_dn = (df_daily['MA10'].iloc[-1] < df_daily['MA10'].iloc[-2] < df_daily['MA10'].iloc[-3])
ma20_dn = (df_daily['MA20'].iloc[-1] < df_daily['MA20'].iloc[-2] < df_daily['MA20'].iloc[-3])
ma_gap = abs(df_daily['MA10'].iloc[-1] - df_daily['MA20'].iloc[-1]) / df_daily['MA20'].iloc[-1] * 100

获取60分钟数据(进场信号)

df_60min = get_price(symbol, end_date=context.current_dt,
frequency='60m', count=50, fields=['close', 'high', 'low', 'open', 'volume'])
if len(df_60min) < 20:
continue

df_60min['MA10'] = df_60min['close'].rolling(context.N1).mean()
df_60min['MA20'] = df_60min['close'].rolling(context.N2).mean()
atr_60min = ATR(df_60min, context.ATR_N)

生成进场信号

做多信号

long_signal = (df_daily['MA10'].iloc[-1] > df_daily['MA20'].iloc[-1]
and ma10_up and ma20_up
and df_daily['close'].iloc[-1] > df_daily['MA10'].iloc[-1]
and ma_gap >= 2
and df_60min['MA10'].iloc[-1] > df_60min['MA20'].iloc[-1]
and df_60min['MA10'].iloc[-1] > df_60min['MA10'].iloc[-2]
and df_60min['MA20'].iloc[-1] > df_60min['MA20'].iloc[-2]
and abs(df_60min['close'].iloc[-1] - df_60min['MA10'].iloc[-1]) <= atr_60min * 1
and ( (df_60min['low'].iloc[-1] - min(df_60min['close'].iloc[-1], df_60min['open'].iloc[-1]) >= 2 * abs(df_60min['close'].iloc[-1] - df_60min['open'].iloc[-1]))
or (df_60min['close'].iloc[-1] > df_60min['close'].iloc[-2] and df_60min['close'].iloc[-2] < df_60min['open'].iloc[-2]) ) )

做空信号

short_signal = (df_daily['MA10'].iloc[-1] < df_daily['MA20'].iloc[-1]
and ma10_dn and ma20_dn
and df_daily['close'].iloc[-1] < df_daily['MA10'].iloc[-1]
and ma_gap >= 2
and df_60min['MA10'].iloc[-1] < df_60min['MA20'].iloc[-1]
and df_60min['MA10'].iloc[-1] < df_60min['MA10'].iloc[-2]
and df_60min['MA20'].iloc[-1] < df_60min['MA20'].iloc[-2]
and abs(df_60min['close'].iloc[-1] - df_60min['MA10'].iloc[-1]) <= atr_60min * 1
and ( (max(df_60min['close'].iloc[-1], df_60min['open'].iloc[-1]) - df_60min['high'].iloc[-1] >= 2 * abs(df_60min['close'].iloc[-1] - df_60min['open'].iloc[-1]))
or (df_60min['close'].iloc[-1] < df_60min['close'].iloc[-2] and df_60min['close'].iloc[-2] > df_60min['open'].iloc[-2]) ) )

计算仓位(以损定仓)

stop_loss = atr_60min * context.ATR_SL
contract_multiplier = get_contract_multiplier(symbol)
position_size = (context.portfolio.portfolio_value * context.risk_ratio) / (stop_loss * contract_multiplier)

执行交易

if long_signal and get_position(symbol).amount == 0:
order_target_percent(symbol, position_size / context.portfolio.portfolio_value)
order_stop_loss(symbol, stop_price=df_60min['close'].iloc[-1] - stop_loss)
order_take_profit(symbol, profit_price=df_60min['close'].iloc[-1] + atr_60min * context.ATR_TP)

if short_signal and get_position(symbol).amount == 0:
order_target_percent(symbol, -position_size / context.portfolio.portfolio_value)
order_stop_loss(symbol, stop_price=df_60min['close'].iloc[-1] + stop_loss)
order_take_profit(symbol, profit_price=df_60min['close'].iloc[-1] - atr_60min * context.ATR_TP)

评论
用户头像
2024-03-03 11:04:20

双均线+ATR交易系统(适配2025年回测时间)

def initialize(context):

系统参数配置

context.N1 = 10 # 短期均线周期
context.N2 = 20 # 长期均线周期
context.ATR_N = 14 # ATR计算周期
context.ATR_SL = 2 # 止损倍数(ATR)
context.ATR_TP = 6 # 止盈倍数(ATR)
context.risk_ratio = 0.01 # 单笔风险比例(1%)

【关键】订阅2025年的有效主力合约

context.symbols = ['SC2503', 'CU2503', 'J2503', 'P2503']
for sym in context.symbols:
subscribe(sym)

ATR指标计算函数

def ATR(df, n):
df['tr'] = df.apply(lambda row: max(row['high'] - row['low'],
abs(row['high'] - row['close'].shift(1)),
abs(row['low'] - row['close'].shift(1))), axis=1)
df['atr'] = df['tr'].rolling(n).mean()
return df['atr'].iloc[-1]

核心交易逻辑

def handle_data(context, data):
for symbol in context.symbols:

获取日线数据(趋势判定)

df_daily = get_price(symbol, end_date=context.current_dt,
frequency='daily', count=50, fields=['close', 'high', 'low'])
if len(df_daily) < 20:
continue

计算日线均线与ATR

df_daily['MA10'] = df_daily['close'].rolling(context.N1).mean()
df_daily['MA20'] = df_daily['close'].rolling(context.N2).mean()
atr_daily = ATR(df_daily, context.ATR_N)

判定日线趋势

ma10_up = (df_daily['MA10'].iloc[-1] > df_daily['MA10'].iloc[-2] > df_daily['MA10'].iloc[-3])
ma20_up = (df_daily['MA20'].iloc[-1] > df_daily['MA20'].iloc[-2] > df_daily['MA20'].iloc[-3])
ma10_dn = (df_daily['MA10'].iloc[-1] < df_daily['MA10'].iloc[-2] < df_daily['MA10'].iloc[-3])
ma20_dn = (df_daily['MA20'].iloc[-1] < df_daily['MA20'].iloc[-2] < df_daily['MA20'].iloc[-3])
ma_gap = abs(df_daily['MA10'].iloc[-1] - df_daily['MA20'].iloc[-1]) / df_daily['MA20'].iloc[-1] * 100

获取60分钟数据(进场信号)

df_60min = get_price(symbol, end_date=context.current_dt,
frequency='60m', count=50, fields=['close', 'high', 'low', 'open', 'volume'])
if len(df_60min) < 20:
continue

df_60min['MA10'] = df_60min['close'].rolling(context.N1).mean()
df_60min['MA20'] = df_60min['close'].rolling(context.N2).mean()
atr_60min = ATR(df_60min, context.ATR_N)

生成进场信号

做多信号

long_signal = (df_daily['MA10'].iloc[-1] > df_daily['MA20'].iloc[-1]
and ma10_up and ma20_up
and df_daily['close'].iloc[-1] > df_daily['MA10'].iloc[-1]
and ma_gap >= 2
and df_60min['MA10'].iloc[-1] > df_60min['MA20'].iloc[-1]
and df_60min['MA10'].iloc[-1] > df_60min['MA10'].iloc[-2]
and df_60min['MA20'].iloc[-1] > df_60min['MA20'].iloc[-2]
and abs(df_60min['close'].iloc[-1] - df_60min['MA10'].iloc[-1]) <= atr_60min * 1
and ( (df_60min['low'].iloc[-1] - min(df_60min['close'].iloc[-1], df_60min['open'].iloc[-1]) >= 2 * abs(df_60min['close'].iloc[-1] - df_60min['open'].iloc[-1]))
or (df_60min['close'].iloc[-1] > df_60min['close'].iloc[-2] and df_60min['close'].iloc[-2] < df_60min['open'].iloc[-2]) ) )

做空信号

short_signal = (df_daily['MA10'].iloc[-1] < df_daily['MA20'].iloc[-1]
and ma10_dn and ma20_dn
and df_daily['close'].iloc[-1] < df_daily['MA10'].iloc[-1]
and ma_gap >= 2
and df_60min['MA10'].iloc[-1] < df_60min['MA20'].iloc[-1]
and df_60min['MA10'].iloc[-1] < df_60min['MA10'].iloc[-2]
and df_60min['MA20'].iloc[-1] < df_60min['MA20'].iloc[-2]
and abs(df_60min['close'].iloc[-1] - df_60min['MA10'].iloc[-1]) <= atr_60min * 1
and ( (max(df_60min['close'].iloc[-1], df_60min['open'].iloc[-1]) - df_60min['high'].iloc[-1] >= 2 * abs(df_60min['close'].iloc[-1] - df_60min['open'].iloc[-1]))
or (df_60min['close'].iloc[-1] < df_60min['close'].iloc[-2] and df_60min['close'].iloc[-2] > df_60min['open'].iloc[-2]) ) )

计算仓位(以损定仓)

stop_loss = atr_60min * context.ATR_SL
contract_multiplier = get_contract_multiplier(symbol)
position_size = (context.portfolio.portfolio_value * context.risk_ratio) / (stop_loss * contract_multiplier)

执行交易

if long_signal and get_position(symbol).amount == 0:
order_target_percent(symbol, position_size / context.portfolio.portfolio_value)
order_stop_loss(symbol, stop_price=df_60min['close'].iloc[-1] - stop_loss)
order_take_profit(symbol, profit_price=df_60min['close'].iloc[-1] + atr_60min * context.ATR_TP)

if short_signal and get_position(symbol).amount == 0:
order_target_percent(symbol, -position_size / context.portfolio.portfolio_value)
order_stop_loss(symbol, stop_price=df_60min['close'].iloc[-1] + stop_loss)
order_take_profit(symbol, profit_price=df_60min['close'].iloc[-1] - atr_60min * context.ATR_TP)

评论
用户头像
2024-04-10 14:19:01

双均线+ATR交易系统(适配2025年回测时间)

def initialize(context):

系统参数配置

context.N1 = 10 # 短期均线周期
context.N2 = 20 # 长期均线周期
context.ATR_N = 14 # ATR计算周期
context.ATR_SL = 2 # 止损倍数(ATR)
context.ATR_TP = 6 # 止盈倍数(ATR)
context.risk_ratio = 0.01 # 单笔风险比例(1%)

【关键】订阅2025年的有效主力合约

context.symbols = ['SC2503', 'CU2503', 'J2503', 'P2503']
for sym in context.symbols:
subscribe(sym)

ATR指标计算函数

def ATR(df, n):
df['tr'] = df.apply(lambda row: max(row['high'] - row['low'],
abs(row['high'] - row['close'].shift(1)),
abs(row['low'] - row['close'].shift(1))), axis=1)
df['atr'] = df['tr'].rolling(n).mean()
return df['atr'].iloc[-1]

核心交易逻辑

def handle_data(context, data):
for symbol in context.symbols:

获取日线数据(趋势判定)

df_daily = get_price(symbol, end_date=context.current_dt,
frequency='daily', count=50, fields=['close', 'high', 'low'])
if len(df_daily) < 20:
continue

计算日线均线与ATR

df_daily['MA10'] = df_daily['close'].rolling(context.N1).mean()
df_daily['MA20'] = df_daily['close'].rolling(context.N2).mean()
atr_daily = ATR(df_daily, context.ATR_N)

判定日线趋势

ma10_up = (df_daily['MA10'].iloc[-1] > df_daily['MA10'].iloc[-2] > df_daily['MA10'].iloc[-3])
ma20_up = (df_daily['MA20'].iloc[-1] > df_daily['MA20'].iloc[-2] > df_daily['MA20'].iloc[-3])
ma10_dn = (df_daily['MA10'].iloc[-1] < df_daily['MA10'].iloc[-2] < df_daily['MA10'].iloc[-3])
ma20_dn = (df_daily['MA20'].iloc[-1] < df_daily['MA20'].iloc[-2] < df_daily['MA20'].iloc[-3])
ma_gap = abs(df_daily['MA10'].iloc[-1] - df_daily['MA20'].iloc[-1]) / df_daily['MA20'].iloc[-1] * 100

获取60分钟数据(进场信号)

df_60min = get_price(symbol, end_date=context.current_dt,
frequency='60m', count=50, fields=['close', 'high', 'low', 'open', 'volume'])
if len(df_60min) < 20:
continue

df_60min['MA10'] = df_60min['close'].rolling(context.N1).mean()
df_60min['MA20'] = df_60min['close'].rolling(context.N2).mean()
atr_60min = ATR(df_60min, context.ATR_N)

生成进场信号

做多信号

long_signal = (df_daily['MA10'].iloc[-1] > df_daily['MA20'].iloc[-1]
and ma10_up and ma20_up
and df_daily['close'].iloc[-1] > df_daily['MA10'].iloc[-1]
and ma_gap >= 2
and df_60min['MA10'].iloc[-1] > df_60min['MA20'].iloc[-1]
and df_60min['MA10'].iloc[-1] > df_60min['MA10'].iloc[-2]
and df_60min['MA20'].iloc[-1] > df_60min['MA20'].iloc[-2]
and abs(df_60min['close'].iloc[-1] - df_60min['MA10'].iloc[-1]) <= atr_60min * 1
and ( (df_60min['low'].iloc[-1] - min(df_60min['close'].iloc[-1], df_60min['open'].iloc[-1]) >= 2 * abs(df_60min['close'].iloc[-1] - df_60min['open'].iloc[-1]))
or (df_60min['close'].iloc[-1] > df_60min['close'].iloc[-2] and df_60min['close'].iloc[-2] < df_60min['open'].iloc[-2]) ) )

做空信号

short_signal = (df_daily['MA10'].iloc[-1] < df_daily['MA20'].iloc[-1]
and ma10_dn and ma20_dn
and df_daily['close'].iloc[-1] < df_daily['MA10'].iloc[-1]
and ma_gap >= 2
and df_60min['MA10'].iloc[-1] < df_60min['MA20'].iloc[-1]
and df_60min['MA10'].iloc[-1] < df_60min['MA10'].iloc[-2]
and df_60min['MA20'].iloc[-1] < df_60min['MA20'].iloc[-2]
and abs(df_60min['close'].iloc[-1] - df_60min['MA10'].iloc[-1]) <= atr_60min * 1
and ( (max(df_60min['close'].iloc[-1], df_60min['open'].iloc[-1]) - df_60min['high'].iloc[-1] >= 2 * abs(df_60min['close'].iloc[-1] - df_60min['open'].iloc[-1]))
or (df_60min['close'].iloc[-1] < df_60min['close'].iloc[-2] and df_60min['close'].iloc[-2] > df_60min['open'].iloc[-2]) ) )

计算仓位(以损定仓)

stop_loss = atr_60min * context.ATR_SL
contract_multiplier = get_contract_multiplier(symbol)
position_size = (context.portfolio.portfolio_value * context.risk_ratio) / (stop_loss * contract_multiplier)

执行交易

if long_signal and get_position(symbol).amount == 0:
order_target_percent(symbol, position_size / context.portfolio.portfolio_value)
order_stop_loss(symbol, stop_price=df_60min['close'].iloc[-1] - stop_loss)
order_take_profit(symbol, profit_price=df_60min['close'].iloc[-1] + atr_60min * context.ATR_TP)

if short_signal and get_position(symbol).amount == 0:
order_target_percent(symbol, -position_size / context.portfolio.portfolio_value)
order_stop_loss(symbol, stop_price=df_60min['close'].iloc[-1] + stop_loss)
order_take_profit(symbol, profit_price=df_60min['close'].iloc[-1] - atr_60min * context.ATR_TP)

评论
用户头像
2025-04-01 10:33:23

双均线+ATR交易系统(适配2025年回测时间)

def initialize(context):

系统参数配置

context.N1 = 10 # 短期均线周期
context.N2 = 20 # 长期均线周期
context.ATR_N = 14 # ATR计算周期
context.ATR_SL = 2 # 止损倍数(ATR)
context.ATR_TP = 6 # 止盈倍数(ATR)
context.risk_ratio = 0.01 # 单笔风险比例(1%)

【关键】订阅2025年的有效主力合约

context.symbols = ['SC2503', 'CU2503', 'J2503', 'P2503']
for sym in context.symbols:
subscribe(sym)

ATR指标计算函数

def ATR(df, n):
df['tr'] = df.apply(lambda row: max(row['high'] - row['low'],
abs(row['high'] - row['close'].shift(1)),
abs(row['low'] - row['close'].shift(1))), axis=1)
df['atr'] = df['tr'].rolling(n).mean()
return df['atr'].iloc[-1]

核心交易逻辑

def handle_data(context, data):
for symbol in context.symbols:

获取日线数据(趋势判定)

df_daily = get_price(symbol, end_date=context.current_dt,
frequency='daily', count=50, fields=['close', 'high', 'low'])
if len(df_daily) < 20:
continue

计算日线均线与ATR

df_daily['MA10'] = df_daily['close'].rolling(context.N1).mean()
df_daily['MA20'] = df_daily['close'].rolling(context.N2).mean()
atr_daily = ATR(df_daily, context.ATR_N)

判定日线趋势

ma10_up = (df_daily['MA10'].iloc[-1] > df_daily['MA10'].iloc[-2] > df_daily['MA10'].iloc[-3])
ma20_up = (df_daily['MA20'].iloc[-1] > df_daily['MA20'].iloc[-2] > df_daily['MA20'].iloc[-3])
ma10_dn = (df_daily['MA10'].iloc[-1] < df_daily['MA10'].iloc[-2] < df_daily['MA10'].iloc[-3])
ma20_dn = (df_daily['MA20'].iloc[-1] < df_daily['MA20'].iloc[-2] < df_daily['MA20'].iloc[-3])
ma_gap = abs(df_daily['MA10'].iloc[-1] - df_daily['MA20'].iloc[-1]) / df_daily['MA20'].iloc[-1] * 100

获取60分钟数据(进场信号)

df_60min = get_price(symbol, end_date=context.current_dt,
frequency='60m', count=50, fields=['close', 'high', 'low', 'open', 'volume'])
if len(df_60min) < 20:
continue

df_60min['MA10'] = df_60min['close'].rolling(context.N1).mean()
df_60min['MA20'] = df_60min['close'].rolling(context.N2).mean()
atr_60min = ATR(df_60min, context.ATR_N)

生成进场信号

做多信号

long_signal = (df_daily['MA10'].iloc[-1] > df_daily['MA20'].iloc[-1]
and ma10_up and ma20_up
and df_daily['close'].iloc[-1] > df_daily['MA10'].iloc[-1]
and ma_gap >= 2
and df_60min['MA10'].iloc[-1] > df_60min['MA20'].iloc[-1]
and df_60min['MA10'].iloc[-1] > df_60min['MA10'].iloc[-2]
and df_60min['MA20'].iloc[-1] > df_60min['MA20'].iloc[-2]
and abs(df_60min['close'].iloc[-1] - df_60min['MA10'].iloc[-1]) <= atr_60min * 1
and ( (df_60min['low'].iloc[-1] - min(df_60min['close'].iloc[-1], df_60min['open'].iloc[-1]) >= 2 * abs(df_60min['close'].iloc[-1] - df_60min['open'].iloc[-1]))
or (df_60min['close'].iloc[-1] > df_60min['close'].iloc[-2] and df_60min['close'].iloc[-2] < df_60min['open'].iloc[-2]) ) )

做空信号

short_signal = (df_daily['MA10'].iloc[-1] < df_daily['MA20'].iloc[-1]
and ma10_dn and ma20_dn
and df_daily['close'].iloc[-1] < df_daily['MA10'].iloc[-1]
and ma_gap >= 2
and df_60min['MA10'].iloc[-1] < df_60min['MA20'].iloc[-1]
and df_60min['MA10'].iloc[-1] < df_60min['MA10'].iloc[-2]
and df_60min['MA20'].iloc[-1] < df_60min['MA20'].iloc[-2]
and abs(df_60min['close'].iloc[-1] - df_60min['MA10'].iloc[-1]) <= atr_60min * 1
and ( (max(df_60min['close'].iloc[-1], df_60min['open'].iloc[-1]) - df_60min['high'].iloc[-1] >= 2 * abs(df_60min['close'].iloc[-1] - df_60min['open'].iloc[-1]))
or (df_60min['close'].iloc[-1] < df_60min['close'].iloc[-2] and df_60min['close'].iloc[-2] > df_60min['open'].iloc[-2]) ) )

计算仓位(以损定仓)

stop_loss = atr_60min * context.ATR_SL
contract_multiplier = get_contract_multiplier(symbol)
position_size = (context.portfolio.portfolio_value * context.risk_ratio) / (stop_loss * contract_multiplier)

执行交易

if long_signal and get_position(symbol).amount == 0:
order_target_percent(symbol, position_size / context.portfolio.portfolio_value)
order_stop_loss(symbol, stop_price=df_60min['close'].iloc[-1] - stop_loss)
order_take_profit(symbol, profit_price=df_60min['close'].iloc[-1] + atr_60min * context.ATR_TP)

if short_signal and get_position(symbol).amount == 0:
order_target_percent(symbol, -position_size / context.portfolio.portfolio_value)
order_stop_loss(symbol, stop_price=df_60min['close'].iloc[-1] + stop_loss)
order_take_profit(symbol, profit_price=df_60min['close'].iloc[-1] - atr_60min * context.ATR_TP)

评论
用户头像
2025-06-25 16:04:52

双均线+ATR交易系统(适配2025年回测时间)

def initialize(context):

系统参数配置

context.N1 = 10 # 短期均线周期
context.N2 = 20 # 长期均线周期
context.ATR_N = 14 # ATR计算周期
context.ATR_SL = 2 # 止损倍数(ATR)
context.ATR_TP = 6 # 止盈倍数(ATR)
context.risk_ratio = 0.01 # 单笔风险比例(1%)

【关键】订阅2025年的有效主力合约

context.symbols = ['SC2503', 'CU2503', 'J2503', 'P2503']
for sym in context.symbols:
subscribe(sym)

ATR指标计算函数

def ATR(df, n):
df['tr'] = df.apply(lambda row: max(row['high'] - row['low'],
abs(row['high'] - row['close'].shift(1)),
abs(row['low'] - row['close'].shift(1))), axis=1)
df['atr'] = df['tr'].rolling(n).mean()
return df['atr'].iloc[-1]

核心交易逻辑

def handle_data(context, data):
for symbol in context.symbols:

获取日线数据(趋势判定)

df_daily = get_price(symbol, end_date=context.current_dt,
frequency='daily', count=50, fields=['close', 'high', 'low'])
if len(df_daily) < 20:
continue

计算日线均线与ATR

df_daily['MA10'] = df_daily['close'].rolling(context.N1).mean()
df_daily['MA20'] = df_daily['close'].rolling(context.N2).mean()
atr_daily = ATR(df_daily, context.ATR_N)

判定日线趋势

ma10_up = (df_daily['MA10'].iloc[-1] > df_daily['MA10'].iloc[-2] > df_daily['MA10'].iloc[-3])
ma20_up = (df_daily['MA20'].iloc[-1] > df_daily['MA20'].iloc[-2] > df_daily['MA20'].iloc[-3])
ma10_dn = (df_daily['MA10'].iloc[-1] < df_daily['MA10'].iloc[-2] < df_daily['MA10'].iloc[-3])
ma20_dn = (df_daily['MA20'].iloc[-1] < df_daily['MA20'].iloc[-2] < df_daily['MA20'].iloc[-3])
ma_gap = abs(df_daily['MA10'].iloc[-1] - df_daily['MA20'].iloc[-1]) / df_daily['MA20'].iloc[-1] * 100

获取60分钟数据(进场信号)

df_60min = get_price(symbol, end_date=context.current_dt,
frequency='60m', count=50, fields=['close', 'high', 'low', 'open', 'volume'])
if len(df_60min) < 20:
continue

df_60min['MA10'] = df_60min['close'].rolling(context.N1).mean()
df_60min['MA20'] = df_60min['close'].rolling(context.N2).mean()
atr_60min = ATR(df_60min, context.ATR_N)

生成进场信号

做多信号

long_signal = (df_daily['MA10'].iloc[-1] > df_daily['MA20'].iloc[-1]
and ma10_up and ma20_up
and df_daily['close'].iloc[-1] > df_daily['MA10'].iloc[-1]
and ma_gap >= 2
and df_60min['MA10'].iloc[-1] > df_60min['MA20'].iloc[-1]
and df_60min['MA10'].iloc[-1] > df_60min['MA10'].iloc[-2]
and df_60min['MA20'].iloc[-1] > df_60min['MA20'].iloc[-2]
and abs(df_60min['close'].iloc[-1] - df_60min['MA10'].iloc[-1]) <= atr_60min * 1
and ( (df_60min['low'].iloc[-1] - min(df_60min['close'].iloc[-1], df_60min['open'].iloc[-1]) >= 2 * abs(df_60min['close'].iloc[-1] - df_60min['open'].iloc[-1]))
or (df_60min['close'].iloc[-1] > df_60min['close'].iloc[-2] and df_60min['close'].iloc[-2] < df_60min['open'].iloc[-2]) ) )

做空信号

short_signal = (df_daily['MA10'].iloc[-1] < df_daily['MA20'].iloc[-1]
and ma10_dn and ma20_dn
and df_daily['close'].iloc[-1] < df_daily['MA10'].iloc[-1]
and ma_gap >= 2
and df_60min['MA10'].iloc[-1] < df_60min['MA20'].iloc[-1]
and df_60min['MA10'].iloc[-1] < df_60min['MA10'].iloc[-2]
and df_60min['MA20'].iloc[-1] < df_60min['MA20'].iloc[-2]
and abs(df_60min['close'].iloc[-1] - df_60min['MA10'].iloc[-1]) <= atr_60min * 1
and ( (max(df_60min['close'].iloc[-1], df_60min['open'].iloc[-1]) - df_60min['high'].iloc[-1] >= 2 * abs(df_60min['close'].iloc[-1] - df_60min['open'].iloc[-1]))
or (df_60min['close'].iloc[-1] < df_60min['close'].iloc[-2] and df_60min['close'].iloc[-2] > df_60min['open'].iloc[-2]) ) )

计算仓位(以损定仓)

stop_loss = atr_60min * context.ATR_SL
contract_multiplier = get_contract_multiplier(symbol)
position_size = (context.portfolio.portfolio_value * context.risk_ratio) / (stop_loss * contract_multiplier)

执行交易

if long_signal and get_position(symbol).amount == 0:
order_target_percent(symbol, position_size / context.portfolio.portfolio_value)
order_stop_loss(symbol, stop_price=df_60min['close'].iloc[-1] - stop_loss)
order_take_profit(symbol, profit_price=df_60min['close'].iloc[-1] + atr_60min * context.ATR_TP)

if short_signal and get_position(symbol).amount == 0:
order_target_percent(symbol, -position_size / context.portfolio.portfolio_value)
order_stop_loss(symbol, stop_price=df_60min['close'].iloc[-1] + stop_loss)
order_take_profit(symbol, profit_price=df_60min['close'].iloc[-1] - atr_60min * context.ATR_TP)

评论
用户头像
2025-07-01 09:06:41

双均线+ATR交易系统(适配2025年回测时间)

def initialize(context):

系统参数配置

context.N1 = 10 # 短期均线周期
context.N2 = 20 # 长期均线周期
context.ATR_N = 14 # ATR计算周期
context.ATR_SL = 2 # 止损倍数(ATR)
context.ATR_TP = 6 # 止盈倍数(ATR)
context.risk_ratio = 0.01 # 单笔风险比例(1%)

【关键】订阅2025年的有效主力合约

context.symbols = ['SC2503', 'CU2503', 'J2503', 'P2503']
for sym in context.symbols:
subscribe(sym)

ATR指标计算函数

def ATR(df, n):
df['tr'] = df.apply(lambda row: max(row['high'] - row['low'],
abs(row['high'] - row['close'].shift(1)),
abs(row['low'] - row['close'].shift(1))), axis=1)
df['atr'] = df['tr'].rolling(n).mean()
return df['atr'].iloc[-1]

核心交易逻辑

def handle_data(context, data):
for symbol in context.symbols:

获取日线数据(趋势判定)

df_daily = get_price(symbol, end_date=context.current_dt,
frequency='daily', count=50, fields=['close', 'high', 'low'])
if len(df_daily) < 20:
continue

计算日线均线与ATR

df_daily['MA10'] = df_daily['close'].rolling(context.N1).mean()
df_daily['MA20'] = df_daily['close'].rolling(context.N2).mean()
atr_daily = ATR(df_daily, context.ATR_N)

判定日线趋势

ma10_up = (df_daily['MA10'].iloc[-1] > df_daily['MA10'].iloc[-2] > df_daily['MA10'].iloc[-3])
ma20_up = (df_daily['MA20'].iloc[-1] > df_daily['MA20'].iloc[-2] > df_daily['MA20'].iloc[-3])
ma10_dn = (df_daily['MA10'].iloc[-1] < df_daily['MA10'].iloc[-2] < df_daily['MA10'].iloc[-3])
ma20_dn = (df_daily['MA20'].iloc[-1] < df_daily['MA20'].iloc[-2] < df_daily['MA20'].iloc[-3])
ma_gap = abs(df_daily['MA10'].iloc[-1] - df_daily['MA20'].iloc[-1]) / df_daily['MA20'].iloc[-1] * 100

获取60分钟数据(进场信号)

df_60min = get_price(symbol, end_date=context.current_dt,
frequency='60m', count=50, fields=['close', 'high', 'low', 'open', 'volume'])
if len(df_60min) < 20:
continue

df_60min['MA10'] = df_60min['close'].rolling(context.N1).mean()
df_60min['MA20'] = df_60min['close'].rolling(context.N2).mean()
atr_60min = ATR(df_60min, context.ATR_N)

生成进场信号

做多信号

long_signal = (df_daily['MA10'].iloc[-1] > df_daily['MA20'].iloc[-1]
and ma10_up and ma20_up
and df_daily['close'].iloc[-1] > df_daily['MA10'].iloc[-1]
and ma_gap >= 2
and df_60min['MA10'].iloc[-1] > df_60min['MA20'].iloc[-1]
and df_60min['MA10'].iloc[-1] > df_60min['MA10'].iloc[-2]
and df_60min['MA20'].iloc[-1] > df_60min['MA20'].iloc[-2]
and abs(df_60min['close'].iloc[-1] - df_60min['MA10'].iloc[-1]) <= atr_60min * 1
and ( (df_60min['low'].iloc[-1] - min(df_60min['close'].iloc[-1], df_60min['open'].iloc[-1]) >= 2 * abs(df_60min['close'].iloc[-1] - df_60min['open'].iloc[-1]))
or (df_60min['close'].iloc[-1] > df_60min['close'].iloc[-2] and df_60min['close'].iloc[-2] < df_60min['open'].iloc[-2]) ) )

做空信号

short_signal = (df_daily['MA10'].iloc[-1] < df_daily['MA20'].iloc[-1]
and ma10_dn and ma20_dn
and df_daily['close'].iloc[-1] < df_daily['MA10'].iloc[-1]
and ma_gap >= 2
and df_60min['MA10'].iloc[-1] < df_60min['MA20'].iloc[-1]
and df_60min['MA10'].iloc[-1] < df_60min['MA10'].iloc[-2]
and df_60min['MA20'].iloc[-1] < df_60min['MA20'].iloc[-2]
and abs(df_60min['close'].iloc[-1] - df_60min['MA10'].iloc[-1]) <= atr_60min * 1
and ( (max(df_60min['close'].iloc[-1], df_60min['open'].iloc[-1]) - df_60min['high'].iloc[-1] >= 2 * abs(df_60min['close'].iloc[-1] - df_60min['open'].iloc[-1]))
or (df_60min['close'].iloc[-1] < df_60min['close'].iloc[-2] and df_60min['close'].iloc[-2] > df_60min['open'].iloc[-2]) ) )

计算仓位(以损定仓)

stop_loss = atr_60min * context.ATR_SL
contract_multiplier = get_contract_multiplier(symbol)
position_size = (context.portfolio.portfolio_value * context.risk_ratio) / (stop_loss * contract_multiplier)

执行交易

if long_signal and get_position(symbol).amount == 0:
order_target_percent(symbol, position_size / context.portfolio.portfolio_value)
order_stop_loss(symbol, stop_price=df_60min['close'].iloc[-1] - stop_loss)
order_take_profit(symbol, profit_price=df_60min['close'].iloc[-1] + atr_60min * context.ATR_TP)

if short_signal and get_position(symbol).amount == 0:
order_target_percent(symbol, -position_size / context.portfolio.portfolio_value)
order_stop_loss(symbol, stop_price=df_60min['close'].iloc[-1] + stop_loss)
order_take_profit(symbol, profit_price=df_60min['close'].iloc[-1] - atr_60min * context.ATR_TP)

评论
用户头像
2025-07-17 01:06:14

双均线+ATR交易系统(适配2025年回测时间)

def initialize(context):

系统参数配置

context.N1 = 10 # 短期均线周期
context.N2 = 20 # 长期均线周期
context.ATR_N = 14 # ATR计算周期
context.ATR_SL = 2 # 止损倍数(ATR)
context.ATR_TP = 6 # 止盈倍数(ATR)
context.risk_ratio = 0.01 # 单笔风险比例(1%)

【关键】订阅2025年的有效主力合约

context.symbols = ['SC2503', 'CU2503', 'J2503', 'P2503']
for sym in context.symbols:
subscribe(sym)

ATR指标计算函数

def ATR(df, n):
df['tr'] = df.apply(lambda row: max(row['high'] - row['low'],
abs(row['high'] - row['close'].shift(1)),
abs(row['low'] - row['close'].shift(1))), axis=1)
df['atr'] = df['tr'].rolling(n).mean()
return df['atr'].iloc[-1]

核心交易逻辑

def handle_data(context, data):
for symbol in context.symbols:

获取日线数据(趋势判定)

df_daily = get_price(symbol, end_date=context.current_dt,
frequency='daily', count=50, fields=['close', 'high', 'low'])
if len(df_daily) < 20:
continue

计算日线均线与ATR

df_daily['MA10'] = df_daily['close'].rolling(context.N1).mean()
df_daily['MA20'] = df_daily['close'].rolling(context.N2).mean()
atr_daily = ATR(df_daily, context.ATR_N)

判定日线趋势

ma10_up = (df_daily['MA10'].iloc[-1] > df_daily['MA10'].iloc[-2] > df_daily['MA10'].iloc[-3])
ma20_up = (df_daily['MA20'].iloc[-1] > df_daily['MA20'].iloc[-2] > df_daily['MA20'].iloc[-3])
ma10_dn = (df_daily['MA10'].iloc[-1] < df_daily['MA10'].iloc[-2] < df_daily['MA10'].iloc[-3])
ma20_dn = (df_daily['MA20'].iloc[-1] < df_daily['MA20'].iloc[-2] < df_daily['MA20'].iloc[-3])
ma_gap = abs(df_daily['MA10'].iloc[-1] - df_daily['MA20'].iloc[-1]) / df_daily['MA20'].iloc[-1] * 100

获取60分钟数据(进场信号)

df_60min = get_price(symbol, end_date=context.current_dt,
frequency='60m', count=50, fields=['close', 'high', 'low', 'open', 'volume'])
if len(df_60min) < 20:
continue

df_60min['MA10'] = df_60min['close'].rolling(context.N1).mean()
df_60min['MA20'] = df_60min['close'].rolling(context.N2).mean()
atr_60min = ATR(df_60min, context.ATR_N)

生成进场信号

做多信号

long_signal = (df_daily['MA10'].iloc[-1] > df_daily['MA20'].iloc[-1]
and ma10_up and ma20_up
and df_daily['close'].iloc[-1] > df_daily['MA10'].iloc[-1]
and ma_gap >= 2
and df_60min['MA10'].iloc[-1] > df_60min['MA20'].iloc[-1]
and df_60min['MA10'].iloc[-1] > df_60min['MA10'].iloc[-2]
and df_60min['MA20'].iloc[-1] > df_60min['MA20'].iloc[-2]
and abs(df_60min['close'].iloc[-1] - df_60min['MA10'].iloc[-1]) <= atr_60min * 1
and ( (df_60min['low'].iloc[-1] - min(df_60min['close'].iloc[-1], df_60min['open'].iloc[-1]) >= 2 * abs(df_60min['close'].iloc[-1] - df_60min['open'].iloc[-1]))
or (df_60min['close'].iloc[-1] > df_60min['close'].iloc[-2] and df_60min['close'].iloc[-2] < df_60min['open'].iloc[-2]) ) )

做空信号

short_signal = (df_daily['MA10'].iloc[-1] < df_daily['MA20'].iloc[-1]
and ma10_dn and ma20_dn
and df_daily['close'].iloc[-1] < df_daily['MA10'].iloc[-1]
and ma_gap >= 2
and df_60min['MA10'].iloc[-1] < df_60min['MA20'].iloc[-1]
and df_60min['MA10'].iloc[-1] < df_60min['MA10'].iloc[-2]
and df_60min['MA20'].iloc[-1] < df_60min['MA20'].iloc[-2]
and abs(df_60min['close'].iloc[-1] - df_60min['MA10'].iloc[-1]) <= atr_60min * 1
and ( (max(df_60min['close'].iloc[-1], df_60min['open'].iloc[-1]) - df_60min['high'].iloc[-1] >= 2 * abs(df_60min['close'].iloc[-1] - df_60min['open'].iloc[-1]))
or (df_60min['close'].iloc[-1] < df_60min['close'].iloc[-2] and df_60min['close'].iloc[-2] > df_60min['open'].iloc[-2]) ) )

计算仓位(以损定仓)

stop_loss = atr_60min * context.ATR_SL
contract_multiplier = get_contract_multiplier(symbol)
position_size = (context.portfolio.portfolio_value * context.risk_ratio) / (stop_loss * contract_multiplier)

执行交易

if long_signal and get_position(symbol).amount == 0:
order_target_percent(symbol, position_size / context.portfolio.portfolio_value)
order_stop_loss(symbol, stop_price=df_60min['close'].iloc[-1] - stop_loss)
order_take_profit(symbol, profit_price=df_60min['close'].iloc[-1] + atr_60min * context.ATR_TP)

if short_signal and get_position(symbol).amount == 0:
order_target_percent(symbol, -position_size / context.portfolio.portfolio_value)
order_stop_loss(symbol, stop_price=df_60min['close'].iloc[-1] + stop_loss)
order_take_profit(symbol, profit_price=df_60min['close'].iloc[-1] - atr_60min * context.ATR_TP)

评论
用户头像
2025-10-13 17:42:26

双均线+ATR交易系统(适配2025年回测时间)

def initialize(context):

系统参数配置

context.N1 = 10 # 短期均线周期
context.N2 = 20 # 长期均线周期
context.ATR_N = 14 # ATR计算周期
context.ATR_SL = 2 # 止损倍数(ATR)
context.ATR_TP = 6 # 止盈倍数(ATR)
context.risk_ratio = 0.01 # 单笔风险比例(1%)

【关键】订阅2025年的有效主力合约

context.symbols = ['SC2503', 'CU2503', 'J2503', 'P2503']
for sym in context.symbols:
subscribe(sym)

ATR指标计算函数

def ATR(df, n):
df['tr'] = df.apply(lambda row: max(row['high'] - row['low'],
abs(row['high'] - row['close'].shift(1)),
abs(row['low'] - row['close'].shift(1))), axis=1)
df['atr'] = df['tr'].rolling(n).mean()
return df['atr'].iloc[-1]

核心交易逻辑

def handle_data(context, data):
for symbol in context.symbols:

获取日线数据(趋势判定)

df_daily = get_price(symbol, end_date=context.current_dt,
frequency='daily', count=50, fields=['close', 'high', 'low'])
if len(df_daily) < 20:
continue

计算日线均线与ATR

df_daily['MA10'] = df_daily['close'].rolling(context.N1).mean()
df_daily['MA20'] = df_daily['close'].rolling(context.N2).mean()
atr_daily = ATR(df_daily, context.ATR_N)

判定日线趋势

ma10_up = (df_daily['MA10'].iloc[-1] > df_daily['MA10'].iloc[-2] > df_daily['MA10'].iloc[-3])
ma20_up = (df_daily['MA20'].iloc[-1] > df_daily['MA20'].iloc[-2] > df_daily['MA20'].iloc[-3])
ma10_dn = (df_daily['MA10'].iloc[-1] < df_daily['MA10'].iloc[-2] < df_daily['MA10'].iloc[-3])
ma20_dn = (df_daily['MA20'].iloc[-1] < df_daily['MA20'].iloc[-2] < df_daily['MA20'].iloc[-3])
ma_gap = abs(df_daily['MA10'].iloc[-1] - df_daily['MA20'].iloc[-1]) / df_daily['MA20'].iloc[-1] * 100

获取60分钟数据(进场信号)

df_60min = get_price(symbol, end_date=context.current_dt,
frequency='60m', count=50, fields=['close', 'high', 'low', 'open', 'volume'])
if len(df_60min) < 20:
continue

df_60min['MA10'] = df_60min['close'].rolling(context.N1).mean()
df_60min['MA20'] = df_60min['close'].rolling(context.N2).mean()
atr_60min = ATR(df_60min, context.ATR_N)

生成进场信号

做多信号

long_signal = (df_daily['MA10'].iloc[-1] > df_daily['MA20'].iloc[-1]
and ma10_up and ma20_up
and df_daily['close'].iloc[-1] > df_daily['MA10'].iloc[-1]
and ma_gap >= 2
and df_60min['MA10'].iloc[-1] > df_60min['MA20'].iloc[-1]
and df_60min['MA10'].iloc[-1] > df_60min['MA10'].iloc[-2]
and df_60min['MA20'].iloc[-1] > df_60min['MA20'].iloc[-2]
and abs(df_60min['close'].iloc[-1] - df_60min['MA10'].iloc[-1]) <= atr_60min * 1
and ( (df_60min['low'].iloc[-1] - min(df_60min['close'].iloc[-1], df_60min['open'].iloc[-1]) >= 2 * abs(df_60min['close'].iloc[-1] - df_60min['open'].iloc[-1]))
or (df_60min['close'].iloc[-1] > df_60min['close'].iloc[-2] and df_60min['close'].iloc[-2] < df_60min['open'].iloc[-2]) ) )

做空信号

short_signal = (df_daily['MA10'].iloc[-1] < df_daily['MA20'].iloc[-1]
and ma10_dn and ma20_dn
and df_daily['close'].iloc[-1] < df_daily['MA10'].iloc[-1]
and ma_gap >= 2
and df_60min['MA10'].iloc[-1] < df_60min['MA20'].iloc[-1]
and df_60min['MA10'].iloc[-1] < df_60min['MA10'].iloc[-2]
and df_60min['MA20'].iloc[-1] < df_60min['MA20'].iloc[-2]
and abs(df_60min['close'].iloc[-1] - df_60min['MA10'].iloc[-1]) <= atr_60min * 1
and ( (max(df_60min['close'].iloc[-1], df_60min['open'].iloc[-1]) - df_60min['high'].iloc[-1] >= 2 * abs(df_60min['close'].iloc[-1] - df_60min['open'].iloc[-1]))
or (df_60min['close'].iloc[-1] < df_60min['close'].iloc[-2] and df_60min['close'].iloc[-2] > df_60min['open'].iloc[-2]) ) )

计算仓位(以损定仓)

stop_loss = atr_60min * context.ATR_SL
contract_multiplier = get_contract_multiplier(symbol)
position_size = (context.portfolio.portfolio_value * context.risk_ratio) / (stop_loss * contract_multiplier)

执行交易

if long_signal and get_position(symbol).amount == 0:
order_target_percent(symbol, position_size / context.portfolio.portfolio_value)
order_stop_loss(symbol, stop_price=df_60min['close'].iloc[-1] - stop_loss)
order_take_profit(symbol, profit_price=df_60min['close'].iloc[-1] + atr_60min * context.ATR_TP)

if short_signal and get_position(symbol).amount == 0:
order_target_percent(symbol, -position_size / context.portfolio.portfolio_value)
order_stop_loss(symbol, stop_price=df_60min['close'].iloc[-1] + stop_loss)
order_take_profit(symbol, profit_price=df_60min['close'].iloc[-1] - atr_60min * context.ATR_TP)

评论
用户头像
2025-10-15 21:07:41

双均线+ATR交易系统(适配2025年回测时间)

def initialize(context):

系统参数配置

context.N1 = 10 # 短期均线周期
context.N2 = 20 # 长期均线周期
context.ATR_N = 14 # ATR计算周期
context.ATR_SL = 2 # 止损倍数(ATR)
context.ATR_TP = 6 # 止盈倍数(ATR)
context.risk_ratio = 0.01 # 单笔风险比例(1%)

【关键】订阅2025年的有效主力合约

context.symbols = ['SC2503', 'CU2503', 'J2503', 'P2503']
for sym in context.symbols:
subscribe(sym)

ATR指标计算函数

def ATR(df, n):
df['tr'] = df.apply(lambda row: max(row['high'] - row['low'],
abs(row['high'] - row['close'].shift(1)),
abs(row['low'] - row['close'].shift(1))), axis=1)
df['atr'] = df['tr'].rolling(n).mean()
return df['atr'].iloc[-1]

核心交易逻辑

def handle_data(context, data):
for symbol in context.symbols:

获取日线数据(趋势判定)

df_daily = get_price(symbol, end_date=context.current_dt,
frequency='daily', count=50, fields=['close', 'high', 'low'])
if len(df_daily) < 20:
continue

计算日线均线与ATR

df_daily['MA10'] = df_daily['close'].rolling(context.N1).mean()
df_daily['MA20'] = df_daily['close'].rolling(context.N2).mean()
atr_daily = ATR(df_daily, context.ATR_N)

判定日线趋势

ma10_up = (df_daily['MA10'].iloc[-1] > df_daily['MA10'].iloc[-2] > df_daily['MA10'].iloc[-3])
ma20_up = (df_daily['MA20'].iloc[-1] > df_daily['MA20'].iloc[-2] > df_daily['MA20'].iloc[-3])
ma10_dn = (df_daily['MA10'].iloc[-1] < df_daily['MA10'].iloc[-2] < df_daily['MA10'].iloc[-3])
ma20_dn = (df_daily['MA20'].iloc[-1] < df_daily['MA20'].iloc[-2] < df_daily['MA20'].iloc[-3])
ma_gap = abs(df_daily['MA10'].iloc[-1] - df_daily['MA20'].iloc[-1]) / df_daily['MA20'].iloc[-1] * 100

获取60分钟数据(进场信号)

df_60min = get_price(symbol, end_date=context.current_dt,
frequency='60m', count=50, fields=['close', 'high', 'low', 'open', 'volume'])
if len(df_60min) < 20:
continue

df_60min['MA10'] = df_60min['close'].rolling(context.N1).mean()
df_60min['MA20'] = df_60min['close'].rolling(context.N2).mean()
atr_60min = ATR(df_60min, context.ATR_N)

生成进场信号

做多信号

long_signal = (df_daily['MA10'].iloc[-1] > df_daily['MA20'].iloc[-1]
and ma10_up and ma20_up
and df_daily['close'].iloc[-1] > df_daily['MA10'].iloc[-1]
and ma_gap >= 2
and df_60min['MA10'].iloc[-1] > df_60min['MA20'].iloc[-1]
and df_60min['MA10'].iloc[-1] > df_60min['MA10'].iloc[-2]
and df_60min['MA20'].iloc[-1] > df_60min['MA20'].iloc[-2]
and abs(df_60min['close'].iloc[-1] - df_60min['MA10'].iloc[-1]) <= atr_60min * 1
and ( (df_60min['low'].iloc[-1] - min(df_60min['close'].iloc[-1], df_60min['open'].iloc[-1]) >= 2 * abs(df_60min['close'].iloc[-1] - df_60min['open'].iloc[-1]))
or (df_60min['close'].iloc[-1] > df_60min['close'].iloc[-2] and df_60min['close'].iloc[-2] < df_60min['open'].iloc[-2]) ) )

做空信号

short_signal = (df_daily['MA10'].iloc[-1] < df_daily['MA20'].iloc[-1]
and ma10_dn and ma20_dn
and df_daily['close'].iloc[-1] < df_daily['MA10'].iloc[-1]
and ma_gap >= 2
and df_60min['MA10'].iloc[-1] < df_60min['MA20'].iloc[-1]
and df_60min['MA10'].iloc[-1] < df_60min['MA10'].iloc[-2]
and df_60min['MA20'].iloc[-1] < df_60min['MA20'].iloc[-2]
and abs(df_60min['close'].iloc[-1] - df_60min['MA10'].iloc[-1]) <= atr_60min * 1
and ( (max(df_60min['close'].iloc[-1], df_60min['open'].iloc[-1]) - df_60min['high'].iloc[-1] >= 2 * abs(df_60min['close'].iloc[-1] - df_60min['open'].iloc[-1]))
or (df_60min['close'].iloc[-1] < df_60min['close'].iloc[-2] and df_60min['close'].iloc[-2] > df_60min['open'].iloc[-2]) ) )

计算仓位(以损定仓)

stop_loss = atr_60min * context.ATR_SL
contract_multiplier = get_contract_multiplier(symbol)
position_size = (context.portfolio.portfolio_value * context.risk_ratio) / (stop_loss * contract_multiplier)

执行交易

if long_signal and get_position(symbol).amount == 0:
order_target_percent(symbol, position_size / context.portfolio.portfolio_value)
order_stop_loss(symbol, stop_price=df_60min['close'].iloc[-1] - stop_loss)
order_take_profit(symbol, profit_price=df_60min['close'].iloc[-1] + atr_60min * context.ATR_TP)

if short_signal and get_position(symbol).amount == 0:
order_target_percent(symbol, -position_size / context.portfolio.portfolio_value)
order_stop_loss(symbol, stop_price=df_60min['close'].iloc[-1] + stop_loss)
order_take_profit(symbol, profit_price=df_60min['close'].iloc[-1] - atr_60min * context.ATR_TP)

评论
用户头像
2025-12-04 14:03:14

双均线+ATR交易系统(适配2025年回测时间)

def initialize(context):

系统参数配置

context.N1 = 10 # 短期均线周期
context.N2 = 20 # 长期均线周期
context.ATR_N = 14 # ATR计算周期
context.ATR_SL = 2 # 止损倍数(ATR)
context.ATR_TP = 6 # 止盈倍数(ATR)
context.risk_ratio = 0.01 # 单笔风险比例(1%)

【关键】订阅2025年的有效主力合约

context.symbols = ['SC2503', 'CU2503', 'J2503', 'P2503']
for sym in context.symbols:
subscribe(sym)

ATR指标计算函数

def ATR(df, n):
df['tr'] = df.apply(lambda row: max(row['high'] - row['low'],
abs(row['high'] - row['close'].shift(1)),
abs(row['low'] - row['close'].shift(1))), axis=1)
df['atr'] = df['tr'].rolling(n).mean()
return df['atr'].iloc[-1]

核心交易逻辑

def handle_data(context, data):
for symbol in context.symbols:

获取日线数据(趋势判定)

df_daily = get_price(symbol, end_date=context.current_dt,
frequency='daily', count=50, fields=['close', 'high', 'low'])
if len(df_daily) < 20:
continue

计算日线均线与ATR

df_daily['MA10'] = df_daily['close'].rolling(context.N1).mean()
df_daily['MA20'] = df_daily['close'].rolling(context.N2).mean()
atr_daily = ATR(df_daily, context.ATR_N)

判定日线趋势

ma10_up = (df_daily['MA10'].iloc[-1] > df_daily['MA10'].iloc[-2] > df_daily['MA10'].iloc[-3])
ma20_up = (df_daily['MA20'].iloc[-1] > df_daily['MA20'].iloc[-2] > df_daily['MA20'].iloc[-3])
ma10_dn = (df_daily['MA10'].iloc[-1] < df_daily['MA10'].iloc[-2] < df_daily['MA10'].iloc[-3])
ma20_dn = (df_daily['MA20'].iloc[-1] < df_daily['MA20'].iloc[-2] < df_daily['MA20'].iloc[-3])
ma_gap = abs(df_daily['MA10'].iloc[-1] - df_daily['MA20'].iloc[-1]) / df_daily['MA20'].iloc[-1] * 100

获取60分钟数据(进场信号)

df_60min = get_price(symbol, end_date=context.current_dt,
frequency='60m', count=50, fields=['close', 'high', 'low', 'open', 'volume'])
if len(df_60min) < 20:
continue

df_60min['MA10'] = df_60min['close'].rolling(context.N1).mean()
df_60min['MA20'] = df_60min['close'].rolling(context.N2).mean()
atr_60min = ATR(df_60min, context.ATR_N)

生成进场信号

做多信号

long_signal = (df_daily['MA10'].iloc[-1] > df_daily['MA20'].iloc[-1]
and ma10_up and ma20_up
and df_daily['close'].iloc[-1] > df_daily['MA10'].iloc[-1]
and ma_gap >= 2
and df_60min['MA10'].iloc[-1] > df_60min['MA20'].iloc[-1]
and df_60min['MA10'].iloc[-1] > df_60min['MA10'].iloc[-2]
and df_60min['MA20'].iloc[-1] > df_60min['MA20'].iloc[-2]
and abs(df_60min['close'].iloc[-1] - df_60min['MA10'].iloc[-1]) <= atr_60min * 1
and ( (df_60min['low'].iloc[-1] - min(df_60min['close'].iloc[-1], df_60min['open'].iloc[-1]) >= 2 * abs(df_60min['close'].iloc[-1] - df_60min['open'].iloc[-1]))
or (df_60min['close'].iloc[-1] > df_60min['close'].iloc[-2] and df_60min['close'].iloc[-2] < df_60min['open'].iloc[-2]) ) )

做空信号

short_signal = (df_daily['MA10'].iloc[-1] < df_daily['MA20'].iloc[-1]
and ma10_dn and ma20_dn
and df_daily['close'].iloc[-1] < df_daily['MA10'].iloc[-1]
and ma_gap >= 2
and df_60min['MA10'].iloc[-1] < df_60min['MA20'].iloc[-1]
and df_60min['MA10'].iloc[-1] < df_60min['MA10'].iloc[-2]
and df_60min['MA20'].iloc[-1] < df_60min['MA20'].iloc[-2]
and abs(df_60min['close'].iloc[-1] - df_60min['MA10'].iloc[-1]) <= atr_60min * 1
and ( (max(df_60min['close'].iloc[-1], df_60min['open'].iloc[-1]) - df_60min['high'].iloc[-1] >= 2 * abs(df_60min['close'].iloc[-1] - df_60min['open'].iloc[-1]))
or (df_60min['close'].iloc[-1] < df_60min['close'].iloc[-2] and df_60min['close'].iloc[-2] > df_60min['open'].iloc[-2]) ) )

计算仓位(以损定仓)

stop_loss = atr_60min * context.ATR_SL
contract_multiplier = get_contract_multiplier(symbol)
position_size = (context.portfolio.portfolio_value * context.risk_ratio) / (stop_loss * contract_multiplier)

执行交易

if long_signal and get_position(symbol).amount == 0:
order_target_percent(symbol, position_size / context.portfolio.portfolio_value)
order_stop_loss(symbol, stop_price=df_60min['close'].iloc[-1] - stop_loss)
order_take_profit(symbol, profit_price=df_60min['close'].iloc[-1] + atr_60min * context.ATR_TP)

if short_signal and get_position(symbol).amount == 0:
order_target_percent(symbol, -position_size / context.portfolio.portfolio_value)
order_stop_loss(symbol, stop_price=df_60min['close'].iloc[-1] + stop_loss)
order_take_profit(symbol, profit_price=df_60min['close'].iloc[-1] - atr_60min * context.ATR_TP)

评论
用户头像
2025-12-06 18:42:00

双均线+ATR交易系统(适配2025年回测时间)

def initialize(context):

系统参数配置

context.N1 = 10 # 短期均线周期
context.N2 = 20 # 长期均线周期
context.ATR_N = 14 # ATR计算周期
context.ATR_SL = 2 # 止损倍数(ATR)
context.ATR_TP = 6 # 止盈倍数(ATR)
context.risk_ratio = 0.01 # 单笔风险比例(1%)

【关键】订阅2025年的有效主力合约

context.symbols = ['SC2503', 'CU2503', 'J2503', 'P2503']
for sym in context.symbols:
subscribe(sym)

ATR指标计算函数

def ATR(df, n):
df['tr'] = df.apply(lambda row: max(row['high'] - row['low'],
abs(row['high'] - row['close'].shift(1)),
abs(row['low'] - row['close'].shift(1))), axis=1)
df['atr'] = df['tr'].rolling(n).mean()
return df['atr'].iloc[-1]

核心交易逻辑

def handle_data(context, data):
for symbol in context.symbols:

获取日线数据(趋势判定)

df_daily = get_price(symbol, end_date=context.current_dt,
frequency='daily', count=50, fields=['close', 'high', 'low'])
if len(df_daily) < 20:
continue

计算日线均线与ATR

df_daily['MA10'] = df_daily['close'].rolling(context.N1).mean()
df_daily['MA20'] = df_daily['close'].rolling(context.N2).mean()
atr_daily = ATR(df_daily, context.ATR_N)

判定日线趋势

ma10_up = (df_daily['MA10'].iloc[-1] > df_daily['MA10'].iloc[-2] > df_daily['MA10'].iloc[-3])
ma20_up = (df_daily['MA20'].iloc[-1] > df_daily['MA20'].iloc[-2] > df_daily['MA20'].iloc[-3])
ma10_dn = (df_daily['MA10'].iloc[-1] < df_daily['MA10'].iloc[-2] < df_daily['MA10'].iloc[-3])
ma20_dn = (df_daily['MA20'].iloc[-1] < df_daily['MA20'].iloc[-2] < df_daily['MA20'].iloc[-3])
ma_gap = abs(df_daily['MA10'].iloc[-1] - df_daily['MA20'].iloc[-1]) / df_daily['MA20'].iloc[-1] * 100

获取60分钟数据(进场信号)

df_60min = get_price(symbol, end_date=context.current_dt,
frequency='60m', count=50, fields=['close', 'high', 'low', 'open', 'volume'])
if len(df_60min) < 20:
continue

df_60min['MA10'] = df_60min['close'].rolling(context.N1).mean()
df_60min['MA20'] = df_60min['close'].rolling(context.N2).mean()
atr_60min = ATR(df_60min, context.ATR_N)

生成进场信号

做多信号

long_signal = (df_daily['MA10'].iloc[-1] > df_daily['MA20'].iloc[-1]
and ma10_up and ma20_up
and df_daily['close'].iloc[-1] > df_daily['MA10'].iloc[-1]
and ma_gap >= 2
and df_60min['MA10'].iloc[-1] > df_60min['MA20'].iloc[-1]
and df_60min['MA10'].iloc[-1] > df_60min['MA10'].iloc[-2]
and df_60min['MA20'].iloc[-1] > df_60min['MA20'].iloc[-2]
and abs(df_60min['close'].iloc[-1] - df_60min['MA10'].iloc[-1]) <= atr_60min * 1
and ( (df_60min['low'].iloc[-1] - min(df_60min['close'].iloc[-1], df_60min['open'].iloc[-1]) >= 2 * abs(df_60min['close'].iloc[-1] - df_60min['open'].iloc[-1]))
or (df_60min['close'].iloc[-1] > df_60min['close'].iloc[-2] and df_60min['close'].iloc[-2] < df_60min['open'].iloc[-2]) ) )

做空信号

short_signal = (df_daily['MA10'].iloc[-1] < df_daily['MA20'].iloc[-1]
and ma10_dn and ma20_dn
and df_daily['close'].iloc[-1] < df_daily['MA10'].iloc[-1]
and ma_gap >= 2
and df_60min['MA10'].iloc[-1] < df_60min['MA20'].iloc[-1]
and df_60min['MA10'].iloc[-1] < df_60min['MA10'].iloc[-2]
and df_60min['MA20'].iloc[-1] < df_60min['MA20'].iloc[-2]
and abs(df_60min['close'].iloc[-1] - df_60min['MA10'].iloc[-1]) <= atr_60min * 1
and ( (max(df_60min['close'].iloc[-1], df_60min['open'].iloc[-1]) - df_60min['high'].iloc[-1] >= 2 * abs(df_60min['close'].iloc[-1] - df_60min['open'].iloc[-1]))
or (df_60min['close'].iloc[-1] < df_60min['close'].iloc[-2] and df_60min['close'].iloc[-2] > df_60min['open'].iloc[-2]) ) )

计算仓位(以损定仓)

stop_loss = atr_60min * context.ATR_SL
contract_multiplier = get_contract_multiplier(symbol)
position_size = (context.portfolio.portfolio_value * context.risk_ratio) / (stop_loss * contract_multiplier)

执行交易

if long_signal and get_position(symbol).amount == 0:
order_target_percent(symbol, position_size / context.portfolio.portfolio_value)
order_stop_loss(symbol, stop_price=df_60min['close'].iloc[-1] - stop_loss)
order_take_profit(symbol, profit_price=df_60min['close'].iloc[-1] + atr_60min * context.ATR_TP)

if short_signal and get_position(symbol).amount == 0:
order_target_percent(symbol, -position_size / context.portfolio.portfolio_value)
order_stop_loss(symbol, stop_price=df_60min['close'].iloc[-1] + stop_loss)
order_take_profit(symbol, profit_price=df_60min['close'].iloc[-1] - atr_60min * context.ATR_TP)

评论
用户头像
2026-01-31 23:36:25

双均线+ATR交易系统(适配2025年回测时间)

def initialize(context):

系统参数配置

context.N1 = 10 # 短期均线周期
context.N2 = 20 # 长期均线周期
context.ATR_N = 14 # ATR计算周期
context.ATR_SL = 2 # 止损倍数(ATR)
context.ATR_TP = 6 # 止盈倍数(ATR)
context.risk_ratio = 0.01 # 单笔风险比例(1%)

【关键】订阅2025年的有效主力合约

context.symbols = ['SC2503', 'CU2503', 'J2503', 'P2503']
for sym in context.symbols:
subscribe(sym)

ATR指标计算函数

def ATR(df, n):
df['tr'] = df.apply(lambda row: max(row['high'] - row['low'],
abs(row['high'] - row['close'].shift(1)),
abs(row['low'] - row['close'].shift(1))), axis=1)
df['atr'] = df['tr'].rolling(n).mean()
return df['atr'].iloc[-1]

核心交易逻辑

def handle_data(context, data):
for symbol in context.symbols:

获取日线数据(趋势判定)

df_daily = get_price(symbol, end_date=context.current_dt,
frequency='daily', count=50, fields=['close', 'high', 'low'])
if len(df_daily) < 20:
continue

计算日线均线与ATR

df_daily['MA10'] = df_daily['close'].rolling(context.N1).mean()
df_daily['MA20'] = df_daily['close'].rolling(context.N2).mean()
atr_daily = ATR(df_daily, context.ATR_N)

判定日线趋势

ma10_up = (df_daily['MA10'].iloc[-1] > df_daily['MA10'].iloc[-2] > df_daily['MA10'].iloc[-3])
ma20_up = (df_daily['MA20'].iloc[-1] > df_daily['MA20'].iloc[-2] > df_daily['MA20'].iloc[-3])
ma10_dn = (df_daily['MA10'].iloc[-1] < df_daily['MA10'].iloc[-2] < df_daily['MA10'].iloc[-3])
ma20_dn = (df_daily['MA20'].iloc[-1] < df_daily['MA20'].iloc[-2] < df_daily['MA20'].iloc[-3])
ma_gap = abs(df_daily['MA10'].iloc[-1] - df_daily['MA20'].iloc[-1]) / df_daily['MA20'].iloc[-1] * 100

获取60分钟数据(进场信号)

df_60min = get_price(symbol, end_date=context.current_dt,
frequency='60m', count=50, fields=['close', 'high', 'low', 'open', 'volume'])
if len(df_60min) < 20:
continue

df_60min['MA10'] = df_60min['close'].rolling(context.N1).mean()
df_60min['MA20'] = df_60min['close'].rolling(context.N2).mean()
atr_60min = ATR(df_60min, context.ATR_N)

生成进场信号

做多信号

long_signal = (df_daily['MA10'].iloc[-1] > df_daily['MA20'].iloc[-1]
and ma10_up and ma20_up
and df_daily['close'].iloc[-1] > df_daily['MA10'].iloc[-1]
and ma_gap >= 2
and df_60min['MA10'].iloc[-1] > df_60min['MA20'].iloc[-1]
and df_60min['MA10'].iloc[-1] > df_60min['MA10'].iloc[-2]
and df_60min['MA20'].iloc[-1] > df_60min['MA20'].iloc[-2]
and abs(df_60min['close'].iloc[-1] - df_60min['MA10'].iloc[-1]) <= atr_60min * 1
and ( (df_60min['low'].iloc[-1] - min(df_60min['close'].iloc[-1], df_60min['open'].iloc[-1]) >= 2 * abs(df_60min['close'].iloc[-1] - df_60min['open'].iloc[-1]))
or (df_60min['close'].iloc[-1] > df_60min['close'].iloc[-2] and df_60min['close'].iloc[-2] < df_60min['open'].iloc[-2]) ) )

做空信号

short_signal = (df_daily['MA10'].iloc[-1] < df_daily['MA20'].iloc[-1]
and ma10_dn and ma20_dn
and df_daily['close'].iloc[-1] < df_daily['MA10'].iloc[-1]
and ma_gap >= 2
and df_60min['MA10'].iloc[-1] < df_60min['MA20'].iloc[-1]
and df_60min['MA10'].iloc[-1] < df_60min['MA10'].iloc[-2]
and df_60min['MA20'].iloc[-1] < df_60min['MA20'].iloc[-2]
and abs(df_60min['close'].iloc[-1] - df_60min['MA10'].iloc[-1]) <= atr_60min * 1
and ( (max(df_60min['close'].iloc[-1], df_60min['open'].iloc[-1]) - df_60min['high'].iloc[-1] >= 2 * abs(df_60min['close'].iloc[-1] - df_60min['open'].iloc[-1]))
or (df_60min['close'].iloc[-1] < df_60min['close'].iloc[-2] and df_60min['close'].iloc[-2] > df_60min['open'].iloc[-2]) ) )

计算仓位(以损定仓)

stop_loss = atr_60min * context.ATR_SL
contract_multiplier = get_contract_multiplier(symbol)
position_size = (context.portfolio.portfolio_value * context.risk_ratio) / (stop_loss * contract_multiplier)

执行交易

if long_signal and get_position(symbol).amount == 0:
order_target_percent(symbol, position_size / context.portfolio.portfolio_value)
order_stop_loss(symbol, stop_price=df_60min['close'].iloc[-1] - stop_loss)
order_take_profit(symbol, profit_price=df_60min['close'].iloc[-1] + atr_60min * context.ATR_TP)

if short_signal and get_position(symbol).amount == 0:
order_target_percent(symbol, -position_size / context.portfolio.portfolio_value)
order_stop_loss(symbol, stop_price=df_60min['close'].iloc[-1] + stop_loss)
order_take_profit(symbol, profit_price=df_60min['close'].iloc[-1] - atr_60min * context.ATR_TP)

评论