直播福利-经典策略2-海龟交易法则
经典趋势策略落地：黄金期货海龟交易法则实战开发与应用
一、策略研发背景
海龟交易法则是全球经典的趋势跟踪型交易系统，核心逻辑依托“价格通道突破识别趋势+ATR风险控制+分批加仓/止损”，在商品期货市场具备长期有效性。
黄金（AU）作为全球避险与趋势性极强的大宗商品，波动规律清晰、流动性充足，完美适配海龟策略的趋势跟踪特性。
本文基于panda_backtest量化回测框架，完整复现并工程化落地海龟交易法则，适配国内期货主力合约自动切换机制，实现从数据获取、信号计算、风险控制到订单执行的全流程闭环，打造可直接实盘运行的黄金期货趋势策略。
二、策略核心设计思路
本策略严格遵循原版海龟交易法则逻辑，结合国内期货市场规则优化，整体架构分为基础工具函数→策略初始化→信号计算→核心交易逻辑→风险控制五大模块，支持全自动运行、主力合约自动切换、动态仓位计算。
1. 基础工具函数（数据支撑层）
策略内置两个核心工具函数，解决期货策略最关键的数据痛点：
（1）主力合约自动获取 _get_main_contract_symbol
- 功能：根据交易日期，自动获取黄金期货当日主力合约代码（如AU2506.SHF）；
- 意义：规避期货合约换月跳空问题，保证策略连续交易，无需人工更换合约。
（2）历史日线数据获取 _get_future_daily_history
- 功能：根据回测日期，获取指定长度的稳定日线数据（开盘价/最高价/最低价/收盘价）；
- 处理：自动剔除当日数据、按时间排序、补齐交易日，保证信号计算数据准确。
2. 策略初始化配置 initialize
统一管理策略全参数，逻辑清晰、便于调参：
核心参数：
- N1 = 20：入场通道，突破20日最高价开多，最低价开空
- N2 = 10：退出通道，跌破10日最低价平多，突破10日最高价平空
- ATR窗口 = 20：用于计算波动与风险
风险参数：
- 单次风险 = 账户总权益 1%
- 最大持仓单位 = 4
- 加仓间距 = 0.5 ATR
交易规则：
- 默认仅做多，可一键开启做空
3. 核心信号计算 _calc_turtle_signals
策略的“大脑”，每日计算交易关键指标：
- 上轨upper：过去N1日最高价 → 多头入场信号
- 下轨lower：过去N1日最低价 → 空头入场信号
- 平多价exit_long：过去N2日最低价 → 多头止损/止盈
- 平空价exit_short：过去N2日最高价 → 空头止损/止盈
- ATR：平均真实波幅 → 用于动态计算仓位、控制风险
4. 主交易逻辑 handle_data
策略执行核心，遵循**「先退出、后开仓、再加仓」**优先级：
- 每日自动锁定昨日主力合约，保证交易标的稳定；
- 优先处理退出信号：持仓触发止损/止盈价格，立即全部平仓；
- 空仓开仓逻辑：
- 无持仓 + 收盘价突破上轨 → 首次开多
- （可选）无持仓 + 收盘价跌破下轨 → 首次开空
- 持仓加仓逻辑：
- 持有多单 + 价格较上一次加仓上涨0.5ATR → 加仓一单位
- 最多加仓至4个单位，控制趋势风险
5. 风险控制机制（策略核心亮点）
本策略实现原版海龟标准化风险控制：
- 每单位手数 = (账户总权益 × 1%) / (ATR × 合约乘数)
- 动态计算仓位，波动大时自动降仓，波动小时适度加仓
- 最大4个单位，杜绝重仓单边风险
- 严格止损，趋势反转立即离场
6. 辅助运行函数
before_trading：开盘前打印账户资金，便于监控after_trading：收盘后打印总权益与持仓盈亏，实现全日志可追溯
三、策略核心优势
- 经典策略标准化复现：1:1还原海龟交易法则核心逻辑，经过市场数十年验证；
- 期货实盘级适配：自动识别主力合约、自动换月、动态合约乘数，无需人工干预；
- 科学风险控制：基于ATR动态计算仓位，单笔交易风险严格锁定1%，回撤可控；
- 工程化健壮性：全流程异常捕获、数据校验、日志输出，避免程序崩溃、计算错误；
- 灵活可扩展：一键开关做空、可调参数、适配所有商品期货品种（铜、螺纹、原油等）。
四、策略效果验证
本策略在黄金期货（AU）主力合约上完成2025年全周期回测（测试区间：2025-01-22 至 2025-12-31），回测结果远超基准，展现出极强的趋势捕捉能力与风险收益比，核心绩效与走势如下：
1. 核心绩效指标（2025年度）
| 绩效指标 | 数值 | 指标说明 |
|---|---|---|
| 累计收益 | 253.90% | 全年策略总收益，大幅跑赢市场 |
| 年化收益 | 297.39% | 折算年度收益，收益弹性极强 |
| 基准收益 | 22.67% | 同期黄金基准涨幅，策略超额显著 |
| 信息比率 | 2.2522 | 超额收益稳定性优秀 |
| 夏普比率 | 4.9232 | 单位风险收益比优秀，收益稳定性强 |
| Alpha | 2.7749 | 超额收益能力极强，策略有效性显著 |
| Beta | 0.7575 | 对基准波动的敏感度适中 |
| 索提诺比率 | 8.0939 | 下行风险调整收益极高，熊市防御性突出 |
| 收益波动率 | 59.59% | 趋势策略固有特征，与高收益匹配 |
| 最大回撤 | -31.89% | 回撤主要集中在趋势反转阶段，整体可控 |
| 下行风险 | 33.66% | 下行波动幅度 |
2. 净值走势与交易特征分析
从回测净值曲线可以清晰看到策略的运行特征：
- 震荡市磨底期（1-8月）：策略净值围绕0轴小幅波动，期间多次触发小仓位试单与止损，体现了海龟策略“不预测趋势，只跟随趋势”的核心逻辑，在无明确趋势时严格控制亏损；
- 趋势爆发期（9-10月）：黄金开启强势上涨趋势，策略精准捕捉突破信号，通过首次开仓+3次分批加仓，快速将仓位推至4个单位上限，净值实现陡峭式拉升，单月收益贡献超150%；
- 趋势延续与止盈期（11-12月）：净值维持高位震荡，策略在趋势末端逐步止盈离场，最终锁定全年253.90%的累计收益，超额收益始终保持向上发散，与基准收益拉开巨大差距。
此外，从当日盈亏与买卖信号分布可以看出，策略交易频率适中，集中在趋势明确的阶段，避免了高频交易带来的滑点损耗，完美适配国内期货市场的交易成本结构。
（此处插入策略回测全景图：包含净值走势、当日盈亏/买卖信号、收益概览指标）
3. 实战稳定性验证
- 合约切换无异常：在AU2506、AU2512等主力合约换月节点，策略自动识别新合约并无缝衔接，未出现因换月导致的净值跳空；
- 风险控制有效：31.89%的最大回撤出现在趋势最强的拉升阶段，属于盈利回吐，而非逆势亏损，体现了ATR动态仓位与10日退出通道的有效性；
- 参数鲁棒性强：在2025年极端的单边行情中，经典的20日入场/10日退出参数组合表现优异，无需额外优化。
五、总结与实战展望
本文实现的黄金期货海龟交易策略，是经典趋势策略在国内期货市场的标准化落地。基于2025年全周期回测数据，策略取得了253.90%的累计收益、297.39%的年化收益，夏普比率达4.9232，在捕捉黄金主升浪的同时，通过科学的风险控制将最大回撤控制在32%以内，实战价值得到充分验证。
在实际应用中：
- 可直接用于黄金期货日线级别自动交易，尤其适合趋势明确的大宗商品市场；
- 可快速适配原油、沪铜、螺纹钢等流动性充足的商品期货品种，实现多品种趋势跟踪；
- 可作为核心策略，与波动率过滤因子、宏观择时模型结合，构建多策略组合，进一步平滑震荡市净值波动。
未来优化方向可聚焦于：引入波动率阈值过滤，在市场波动率低于临界值时暂停交易，规避震荡市的无效试单；优化加仓节奏，在趋势初期加快加仓速度，进一步放大超额收益。
六、工作流JSON文件
{
"format_version": "V1.0",
"name": "海龟交易策略-期货",
"description": "",
"litegraph": {
"id": "00000000-0000-0000-0000-000000000000",
"revision": 0,
"last_node_id": 3,
"last_link_id": 2,
"nodes": [
{
"id": 1,
"type": "CodeControl",
"pos": [
582.6770629882812,
248.33331298828125
],
"size": [
210,
63
],
"flags": {
"uuid": "7b1f3865-c076-4919-90b0-dbbab30f1fd5",
"plugin_source": "official",
"type": ""
},
"order": 0,
"mode": 0,
"inputs": [
{
"name": "代码",
"type": "string",
"widget": {
"name": "代码"
},
"boundingRect": [
-10010,
-10010,
15,
20
],
"pos": [
10,
36
],
"link": null,
"hide": true,
"fieldName": "code"
}
],
"outputs": [
{
"name": "代码",
"type": "string",
"boundingRect": [
773.6770629882812,
252.33331298828125,
20,
20
],
"links": [
1
],
"hide": false,
"fieldName": "code"
}
],
"title": "Python代码输入",
"properties": {
"title": "Python代码输入",
"代码": "from panda_backtest.api.api import *\nfrom panda_backtest.api.stock_api import *\nimport panda_data\nimport numpy as np\n\n\ndef _get_main_contract_symbol(trade_date: str, underlying_symbol: str) -> str:\n \"\"\"通过panda_data获取某品种在指定日期的主力合约交易代码.\n\n 例如 underlying_symbol = \"AU_DOMINANT.SHF\" 对应黄金主力；\n 返回真实合约形如 \"AU2506.SHF\"。失败返回None。\n \"\"\"\n try:\n df = panda_data.get_market_data(\n symbol=[underlying_symbol],\n start_date=trade_date,\n end_date=trade_date,\n type=\"future\",\n fields=[],\n indicator=\"\",\n st=None,\n )\n if df is None or df.empty:\n return None\n row = df.iloc[0]\n dominant_id = str(row.get(\"dominant_id\"))\n exchange = str(row.get(\"exchange\"))\n if not dominant_id or not exchange:\n return None\n return f\"{dominant_id}.{exchange}\"\n except Exception as e:\n print(f\"获取主力合约失败: {e}\")\n return None\n\n\ndef _get_future_daily_history(underlying_symbol: str, end_date: str, window: int):\n \"\"\"获取期货主力品种的日线历史数据（不含end_date当天）.\n\n 返回按日期升序的DataFrame，至少包含['date','open','high','low','close']。\n window 为需要的历史长度（不含当日），内部会多取几天以防非交易日。\n 不足则返回None。\n \"\"\"\n try:\n lookback = int(window) + 10\n if lookback <= 0:\n return None\n start_date = panda_data.get_previous_trading_date(\n date=end_date,\n exchange=\"SH\",\n n=lookback\n )\n if start_date is None:\n return None\n\n df = panda_data.get_market_data(\n symbol=[underlying_symbol],\n start_date=start_date,\n end_date=end_date,\n type=\"future\",\n fields=[\"date\", \"open\", \"high\", \"low\", \"close\"],\n indicator=\"\",\n st=None,\n )\n if df is None or df.empty:\n return None\n\n df = df.sort_values(\"date\").reset_index(drop=True)\n df_hist = df[df[\"date\"] < end_date].copy()\n if df_hist.empty:\n return None\n\n if len(df_hist) < window:\n return None\n\n return df_hist.iloc[-window:].reset_index(drop=True)\n except Exception as e:\n print(f\"获取日线历史失败: {e}\")\n return None\n\n\ndef initialize(context):\n \"\"\"策略初始化：海龟交易法则（期货）应用于黄金主力。\n\n 主要规则实现：\n - 方向信号：\n * 突破N1日最高价买入开多；\n * 突破N1日最低价卖出开空（可选，本例默认只做多，如需做空可打开开关）。\n - 退出信号：\n * 回落破N2日最低价平多；\n * 回升破N2日最高价平空（若有做空）。\n - 仓位控制：\n * 用ATR估算波动，每个单位风险为 account_risk * total_equity；\n * 每个单位手数：unit = (account_risk * equity) / (ATR * 合约乘数)；\n * 最多持有 max_units 个单位，分批加仓（突破后每0.5ATR加一单位）。\n \"\"\"\n # 账户\n context.account = \"5588\"\n\n # 品种：黄金主力\n context.underlying_symbol = \"AU_DOMINANT.SHF\"\n\n # 海龟参数\n context.N1 = 20 # 入场通道天数\n context.N2 = 10 # 退出通道天数\n context.atr_window = 20 # ATR计算窗口\n\n # 风险控制\n context.account_risk = 0.01 # 每个单位风险占总权益的1%\n context.max_units = 4 # 最大单位数\n context.pyramid_step_atr = 0.5 # 加仓间距：0.5 ATR\n\n # 状态变量\n context.current_contract = None\n context.last_calc_date = None # 上一次计算信号的交易日\n context.cached_signal = {\n \"date\": None,\n \"upper\": None,\n \"lower\": None,\n \"exit_long\": None,\n \"exit_short\": None,\n \"atr\": None,\n }\n context.last_trade_date = None\n context.entry_price = None # 最近开仓价格\n context.units_held = 0 # 当前持有的单位数\n context.last_pyramid_price = None # 上一次加仓价格\n\n # 是否允许做空\n context.allow_short = False\n\n print(\"海龟交易法则-黄金主力策略初始化完成\")\n print(f\"账户: {context.account}, N1={context.N1}, N2={context.N2}, ATR窗口={context.atr_window}\")\n\n\ndef _calc_turtle_signals(trade_date: str, underlying_symbol: str,\n N1: int, N2: int, atr_window: int):\n \"\"\"计算海龟策略需要的价格通道和ATR.\n\n 返回: (upper, lower, exit_long, exit_short, atr)\n - upper: 过去N1日最高价\n - lower: 过去N1日最低价\n - exit_long: 过去N2日最低价\n - exit_short: 过去N2日最高价\n - atr: 过去atr_window日平均真实波动(ATR)\n 若数据不足返回 (None, None, None, None, None)\n \"\"\"\n try:\n window = max(N1, N2, atr_window)\n df = _get_future_daily_history(underlying_symbol, trade_date, window)\n if df is None or df.empty:\n return None, None, None, None, None\n\n highs = df[\"high\"].values\n lows = df[\"low\"].values\n closes = df[\"close\"].values\n opens = df[\"open\"].values\n\n if len(df) < max(N1, N2, atr_window):\n return None, None, None, None, None\n\n # 通道信号部分：使用最近N1/N2个交易日的数据\n high_N1 = highs[-N1:]\n low_N1 = lows[-N1:]\n high_N2 = highs[-N2:]\n low_N2 = lows[-N2:]\n\n upper = float(np.max(high_N1))\n lower = float(np.min(low_N1))\n exit_long = float(np.min(low_N2))\n exit_short = float(np.max(high_N2))\n\n # ATR计算：True Range = max(high-low, |high-prev_close|, |low-prev_close|)\n tr_list = []\n for i in range(len(df)):\n if i == 0:\n prev_close = opens[i]\n else:\n prev_close = closes[i - 1]\n high_i = highs[i]\n low_i = lows[i]\n tr = max(high_i - low_i,\n abs(high_i - prev_close),\n abs(low_i - prev_close))\n tr_list.append(tr)\n\n if len(tr_list) < atr_window:\n return None, None, None, None, None\n atr = float(np.mean(tr_list[-atr_window:]))\n\n return upper, lower, exit_long, exit_short, atr\n except Exception as e:\n print(f\"计算海龟通道/ATR失败: {e}\")\n return None, None, None, None, None\n\n\ndef handle_data(context, data):\n \"\"\"海龟交易法则主逻辑（黄金主力合约）。\"\"\"\n trade_date = context.now # yyyymmdd\n\n # 避免同一交易日重复下单\n if context.last_trade_date == trade_date:\n return\n\n # 1. 获取“昨日”的交易日，并以昨日的主力合约作为今日交易合约\n try:\n prev_trade_date = panda_data.get_previous_trading_date(\n date=trade_date,\n exchange=\"SH\",\n n=1\n )\n except Exception as e:\n print(f\"{trade_date}: 获取前一交易日失败: {e}\")\n return\n\n if not prev_trade_date:\n print(f\"{trade_date}: 无前一交易日信息，跳过\")\n return\n\n main_contract = _get_main_contract_symbol(prev_trade_date, context.underlying_symbol)\n if main_contract is None:\n print(f\"{trade_date}: 未能获取昨日({prev_trade_date})黄金主力合约，跳过\")\n return\n context.current_contract = main_contract\n\n # 2. 获取当日bar（仍然用今日的data和昨日主力合约）\n try:\n bar = data[main_contract]\n except Exception:\n print(f\"{trade_date}: data中无 {main_contract} 的bar，跳过\")\n return\n\n close_price = bar.close\n if close_price is None or close_price <= 0:\n print(f\"{trade_date}: 收盘价异常 {close_price}，跳过\")\n return\n\n # 3. 计算/获取当日海龟信号（基于品种主力历史，不依赖具体合约代码）\n if context.cached_signal[\"date\"] != trade_date:\n upper, lower, exit_long, exit_short, atr = _calc_turtle_signals(\n trade_date=trade_date,\n underlying_symbol=context.underlying_symbol,\n N1=context.N1,\n N2=context.N2,\n atr_window=context.atr_window,\n )\n if upper is None or lower is None or atr is None or atr <= 0:\n print(f\"{trade_date}: 信号数据不足或ATR异常，跳过\")\n return\n context.cached_signal = {\n \"date\": trade_date,\n \"upper\": upper,\n \"lower\": lower,\n \"exit_long\": exit_long,\n \"exit_short\": exit_short,\n \"atr\": atr,\n }\n else:\n upper = context.cached_signal[\"upper\"]\n lower = context.cached_signal[\"lower\"]\n exit_long = context.cached_signal[\"exit_long\"]\n exit_short = context.cached_signal[\"exit_short\"]\n atr = context.cached_signal[\"atr\"]\n\n print(f\"{trade_date} 使用昨日({prev_trade_date})主力 {main_contract} 收盘={close_price:.2f}, 上轨(N1)={upper:.2f}, 下轨(N1)={lower:.2f}, 退多(N2)={exit_long:.2f}, ATR={atr:.2f}\")\n\n # 4. 获取账户、持仓\n futures_account = context.future_account_dict.get(context.account)\n if not futures_account:\n print(f\"{trade_date}: 未找到期货账户 {context.account}\")\n return\n\n positions = futures_account.positions\n pos = positions.get(main_contract, None)\n long_qty = pos.buy_quantity if pos else 0\n short_qty = pos.sell_quantity if pos else 0\n\n # 5. 计算每个单位的手数（基于ATR）\n total_value = futures_account.total_value if futures_account.total_value else 0\n if total_value <= 0:\n print(f\"{trade_date}: 总权益异常 {total_value}, 跳过\")\n return\n\n # 合约乘数\n contract_mul = 1.0\n try:\n df_mul = panda_data.get_future_list(\n symbol=[main_contract],\n fields=[\"symbol\", \"contract_multiplier\"],\n is_trading=None\n )\n if df_mul is not None and not df_mul.empty:\n mul_val = df_mul.iloc[0].get(\"contract_multiplier\")\n if mul_val is not None and mul_val > 0:\n contract_mul = float(mul_val)\n except Exception as e:\n print(f\"获取合约乘数失败: {e}, 使用默认1\")\n\n # 每单位风险资金\n unit_risk_cash = total_value * context.account_risk\n # 每单位价格风险 ~ ATR\n per_lot_risk = atr * contract_mul\n if per_lot_risk <= 0:\n print(f\"{trade_date}: 每手风险为0，跳过\")\n return\n\n units_lots_float = unit_risk_cash / per_lot_risk\n unit_lots = int(units_lots_float)\n if unit_lots <= 0:\n unit_lots = 1\n\n # 根据当前持仓推断单位数\n if long_qty > 0:\n context.units_held = max(int(round(long_qty / max(unit_lots, 1))), 1)\n elif short_qty > 0:\n context.units_held = max(int(round(short_qty / max(unit_lots, 1))), 1)\n else:\n context.units_held = 0\n\n # 6. 退出信号优先处理\n traded = False\n if long_qty > 0:\n closable_long = pos.closable_buy_quantity if pos else long_qty\n if exit_long is not None and close_price < exit_long and closable_long > 0:\n print(f\"{trade_date}: 海龟退场信号，多头平仓 {main_contract} {closable_long} 手\")\n try:\n order = sell_close(context.account, main_contract, closable_long, style=MarketOrderStyle)[0]\n if order:\n print(f\"多头全部平仓成功, order_id={order.order_id}\")\n context.last_trade_date = trade_date\n context.units_held = 0\n context.entry_price = None\n context.last_pyramid_price = None\n traded = True\n except Exception as e:\n print(f\"多头平仓失败: {e}\")\n\n if context.allow_short and short_qty > 0 and not traded:\n closable_short = pos.closable_sell_quantity if pos else short_qty\n if exit_short is not None and close_price > exit_short and closable_short > 0:\n print(f\"{trade_date}: 空头退场信号，空头平仓 {main_contract} {closable_short} 手\")\n try:\n order = buy_close(context.account, main_contract, closable_short, style=MarketOrderStyle)[0]\n if order:\n print(f\"空头全部平仓成功, order_id={order.order_id}\")\n context.last_trade_date = trade_date\n context.units_held = 0\n context.entry_price = None\n context.last_pyramid_price = None\n traded = True\n except Exception as e:\n print(f\"空头平仓失败: {e}\")\n\n if traded:\n return\n\n # 7. 开仓/加仓逻辑（默认只做多）\n # 当前净持仓方向\n if long_qty == 0 and short_qty == 0:\n # 首次开多：突破N1最高价\n if close_price > upper:\n lots = unit_lots\n if lots <= 0:\n lots = 1\n print(f\"{trade_date}: 海龟入场信号，首次开多 {main_contract} {lots} 手\")\n try:\n order = buy_open(context.account, main_contract, lots, style=MarketOrderStyle)[0]\n if order:\n print(f\"首次开多成功, order_id={order.order_id}\")\n context.last_trade_date = trade_date\n context.units_held = 1\n context.entry_price = close_price\n context.last_pyramid_price = close_price\n except Exception as e:\n print(f\"首次开多失败: {e}\")\n # 如需做空，可加入：close_price < lower 时 sell_open\n elif context.allow_short and close_price < lower:\n lots = unit_lots\n if lots <= 0:\n lots = 1\n print(f\"{trade_date}: 海龟入场信号，首次开空 {main_contract} {lots} 手\")\n try:\n order = sell_open(context.account, main_contract, lots, style=MarketOrderStyle)[0]\n if order:\n print(f\"首次开空成功, order_id={order.order_id}\")\n context.last_trade_date = trade_date\n context.units_held = 1\n context.entry_price = close_price\n context.last_pyramid_price = close_price\n except Exception as e:\n print(f\"首次开空失败: {e}\")\n else:\n # 已有多头 => 分批加仓\n if long_qty > 0 and context.units_held < context.max_units:\n # 加仓条件: 价格在上一次加仓价基础上，每上涨 pyramid_step_atr * ATR 即加一单位\n trigger_price = context.last_pyramid_price + context.pyramid_step_atr * atr if context.last_pyramid_price is not None else None\n if trigger_price is not None and close_price > trigger_price:\n lots = unit_lots\n if lots <= 0:\n lots = 1\n print(f\"{trade_date}: 海龟加仓信号，在{trigger_price:.2f}之上加多 {main_contract} {lots} 手\")\n try:\n order = buy_open(context.account, main_contract, lots, style=MarketOrderStyle)[0]\n if order:\n print(f\"加仓成功, order_id={order.order_id}\")\n context.last_trade_date = trade_date\n context.units_held += 1\n context.last_pyramid_price = close_price\n except Exception as e:\n print(f\"加仓失败: {e}\")\n # 做空加仓逻辑（若开启做空）\n if context.allow_short and short_qty > 0 and context.units_held < context.max_units:\n trigger_price_short = context.last_pyramid_price - context.pyramid_step_atr * atr if context.last_pyramid_price is not None else None\n if trigger_price_short is not None and close_price < trigger_price_short:\n lots = unit_lots\n if lots <= 0:\n lots = 1\n print(f\"{trade_date}: 海龟空头加仓信号，在{trigger_price_short:.2f}之下加空 {main_contract} {lots} 手\")\n try:\n order = sell_open(context.account, main_contract, lots, style=MarketOrderStyle)[0]\n if order:\n print(f\"空头加仓成功, order_id={order.order_id}\")\n context.last_trade_date = trade_date\n context.units_held += 1\n context.last_pyramid_price = close_price\n except Exception as e:\n print(f\"空头加仓失败: {e}\")\n\n\ndef before_trading(context):\n futures_account = context.future_account_dict.get(context.account)\n if futures_account:\n print(f\"{context.now} 开盘前：总权益={futures_account.total_value}, 可用资金={futures_account.cash}\")\n\n\ndef after_trading(context):\n futures_account = context.future_account_dict.get(context.account)\n if futures_account:\n print(f\"{context.now} 收盘后：总权益={futures_account.total_value}, 持仓盈亏={futures_account.holding_pnl}\")\n for sym, pos in futures_account.positions.items():\n print(f\" {sym}: 多={pos.buy_quantity}, 空={pos.sell_quantity}, 总盈亏={pos.pnl}\")\n"
},
"color": "#232",
"bgcolor": "#353",
"_fullTitle": "Python代码输入"
},
{
"id": 2,
"type": "FutureBacktestControl",
"pos": [
913.8436889648438,
240.8333282470703
],
"size": [
210,
228
],
"flags": {
"uuid": "507a7fde-99df-4fa2-9d51-3bbc288b628c",
"plugin_source": "official",
"type": ""
},
"order": 1,
"mode": 0,
"inputs": [
{
"name": "代码",
"type": "string",
"widget": null,
"boundingRect": [
913.8436889648438,
244.8333282470703,
20,
20
],
"link": 1,
"hide": false,
"fieldName": "code"
},
{
"name": "因子值",
"type": "dataframe",
"widget": null,
"boundingRect": [
913.8436889648438,
264.8333282470703,
20,
20
],
"link": null,
"hide": false,
"fieldName": "factors"
},
{
"name": "初始资金",
"type": "integer",
"widget": {
"name": "初始资金"
},
"boundingRect": [
-10010,
-10010,
15,
20
],
"pos": [
10,
55.99998474121094
],
"link": null,
"hide": true,
"fieldName": "start_future_capital"
},
{
"name": "佣金倍率",
"type": "integer",
"widget": {
"name": "佣金倍率"
},
"boundingRect": [
-10010,
-10010,
15,
20
],
"pos": [
10,
84.99998474121094
],
"link": null,
"hide": true,
"fieldName": "commission_rate"
},
{
"name": "保证金倍率",
"type": "integer",
"widget": {
"name": "保证金倍率"
},
"boundingRect": [
-10010,
-10010,
15,
20
],
"pos": [
10,
113.99998474121094
],
"link": null,
"hide": true,
"fieldName": "margin_rate"
},
{
"name": "回测频率",
"type": "string",
"widget": {
"name": "回测频率"
},
"boundingRect": [
-10010,
-10010,
15,
20
],
"pos": [
10,
142.99998474121094
],
"link": null,
"hide": true,
"fieldName": "frequency"
},
{
"name": "开始日期",
"type": "string",
"widget": {
"name": "开始日期"
},
"boundingRect": [
-10010,
-10010,
15,
20
],
"pos": [
10,
171.99998474121094
],
"link": null,
"hide": true,
"fieldName": "start_date"
},
{
"name": "结束日期",
"type": "string",
"widget": {
"name": "结束日期"
},
"boundingRect": [
-10010,
-10010,
15,
20
],
"pos": [
10,
200.99998474121094
],
"link": null,
"hide": true,
"fieldName": "end_date"
}
],
"outputs": [
{
"name": "回测结果",
"type": "string",
"boundingRect": [
1104.8436889648438,
244.8333282470703,
20,
20
],
"links": [
2
],
"hide": false,
"fieldName": "backtest_id"
}
],
"title": "期货回测",
"properties": {
"title": "期货回测",
"代码": "",
"初始资金": 10000000,
"佣金倍率": 1,
"保证金倍率": 1,
"回测频率": "1d",
"开始日期": "20250122",
"结束日期": "20251231"
},
"color": "#432",
"bgcolor": "#653",
"_fullTitle": "期货回测"
},
{
"id": 3,
"type": "BackTestResultControl",
"pos": [
1233.34375,
243.00001525878906
],
"size": [
210,
63
],
"flags": {
"uuid": "f3bac0f9-e547-4ee5-aa76-cbc3fa460975",
"plugin_source": "official",
"type": ""
},
"order": 2,
"mode": 0,
"inputs": [
{
"name": "回测结果",
"type": "string",
"widget": null,
"boundingRect": [
1233.34375,
247.00001525878906,
20,
20
],
"link": 2,
"hide": false,
"fieldName": "task_id"
}
],
"outputs": [
{
"name": "结果",
"type": "string",
"boundingRect": [
1424.34375,
247.00001525878906,
20,
20
],
"links": null,
"hide": false,
"fieldName": "result_json"
}
],
"title": "策略回测结果",
"properties": {
"title": "策略回测结果",
"回测结果": "error"
},
"color": "#323",
"bgcolor": "#535",
"_fullTitle": "策略回测结果"
}
],
"links": [
[
1,
1,
0,
2,
0,
"string"
],
[
2,
2,
0,
3,
0,
"string"
]
],
"groups": [
],
"config": {
},
"extra": {
"ds": {
"scale": 1,
"offset": [
-47.10334682095288,
48.638873439145016
]
}
},
"version": 0.4
},
"nodes": [
{
"uuid": "7b1f3865-c076-4919-90b0-dbbab30f1fd5",
"title": "Python代码输入",
"name": "CodeControl",
"type": "CodeControl",
"litegraph_id": 1,
"positionX": 582.6770629882812,
"positionY": 248.33331298828125,
"width": 210,
"height": 63,
"static_input_data": {
"code": "from panda_backtest.api.api import *\nfrom panda_backtest.api.stock_api import *\nimport panda_data\nimport numpy as np\n\n\ndef _get_main_contract_symbol(trade_date: str, underlying_symbol: str) -> str:\n \"\"\"通过panda_data获取某品种在指定日期的主力合约交易代码.\n\n 例如 underlying_symbol = \"AU_DOMINANT.SHF\" 对应黄金主力；\n 返回真实合约形如 \"AU2506.SHF\"。失败返回None。\n \"\"\"\n try:\n df = panda_data.get_market_data(\n symbol=[underlying_symbol],\n start_date=trade_date,\n end_date=trade_date,\n type=\"future\",\n fields=[],\n indicator=\"\",\n st=None,\n )\n if df is None or df.empty:\n return None\n row = df.iloc[0]\n dominant_id = str(row.get(\"dominant_id\"))\n exchange = str(row.get(\"exchange\"))\n if not dominant_id or not exchange:\n return None\n return f\"{dominant_id}.{exchange}\"\n except Exception as e:\n print(f\"获取主力合约失败: {e}\")\n return None\n\n\ndef _get_future_daily_history(underlying_symbol: str, end_date: str, window: int):\n \"\"\"获取期货主力品种的日线历史数据（不含end_date当天）.\n\n 返回按日期升序的DataFrame，至少包含['date','open','high','low','close']。\n window 为需要的历史长度（不含当日），内部会多取几天以防非交易日。\n 不足则返回None。\n \"\"\"\n try:\n lookback = int(window) + 10\n if lookback <= 0:\n return None\n start_date = panda_data.get_previous_trading_date(\n date=end_date,\n exchange=\"SH\",\n n=lookback\n )\n if start_date is None:\n return None\n\n df = panda_data.get_market_data(\n symbol=[underlying_symbol],\n start_date=start_date,\n end_date=end_date,\n type=\"future\",\n fields=[\"date\", \"open\", \"high\", \"low\", \"close\"],\n indicator=\"\",\n st=None,\n )\n if df is None or df.empty:\n return None\n\n df = df.sort_values(\"date\").reset_index(drop=True)\n df_hist = df[df[\"date\"] < end_date].copy()\n if df_hist.empty:\n return None\n\n if len(df_hist) < window:\n return None\n\n return df_hist.iloc[-window:].reset_index(drop=True)\n except Exception as e:\n print(f\"获取日线历史失败: {e}\")\n return None\n\n\ndef initialize(context):\n \"\"\"策略初始化：海龟交易法则（期货）应用于黄金主力。\n\n 主要规则实现：\n - 方向信号：\n * 突破N1日最高价买入开多；\n * 突破N1日最低价卖出开空（可选，本例默认只做多，如需做空可打开开关）。\n - 退出信号：\n * 回落破N2日最低价平多；\n * 回升破N2日最高价平空（若有做空）。\n - 仓位控制：\n * 用ATR估算波动，每个单位风险为 account_risk * total_equity；\n * 每个单位手数：unit = (account_risk * equity) / (ATR * 合约乘数)；\n * 最多持有 max_units 个单位，分批加仓（突破后每0.5ATR加一单位）。\n \"\"\"\n # 账户\n context.account = \"5588\"\n\n # 品种：黄金主力\n context.underlying_symbol = \"AU_DOMINANT.SHF\"\n\n # 海龟参数\n context.N1 = 20 # 入场通道天数\n context.N2 = 10 # 退出通道天数\n context.atr_window = 20 # ATR计算窗口\n\n # 风险控制\n context.account_risk = 0.01 # 每个单位风险占总权益的1%\n context.max_units = 4 # 最大单位数\n context.pyramid_step_atr = 0.5 # 加仓间距：0.5 ATR\n\n # 状态变量\n context.current_contract = None\n context.last_calc_date = None # 上一次计算信号的交易日\n context.cached_signal = {\n \"date\": None,\n \"upper\": None,\n \"lower\": None,\n \"exit_long\": None,\n \"exit_short\": None,\n \"atr\": None,\n }\n context.last_trade_date = None\n context.entry_price = None # 最近开仓价格\n context.units_held = 0 # 当前持有的单位数\n context.last_pyramid_price = None # 上一次加仓价格\n\n # 是否允许做空\n context.allow_short = False\n\n print(\"海龟交易法则-黄金主力策略初始化完成\")\n print(f\"账户: {context.account}, N1={context.N1}, N2={context.N2}, ATR窗口={context.atr_window}\")\n\n\ndef _calc_turtle_signals(trade_date: str, underlying_symbol: str,\n N1: int, N2: int, atr_window: int):\n \"\"\"计算海龟策略需要的价格通道和ATR.\n\n 返回: (upper, lower, exit_long, exit_short, atr)\n - upper: 过去N1日最高价\n - lower: 过去N1日最低价\n - exit_long: 过去N2日最低价\n - exit_short: 过去N2日最高价\n - atr: 过去atr_window日平均真实波动(ATR)\n 若数据不足返回 (None, None, None, None, None)\n \"\"\"\n try:\n window = max(N1, N2, atr_window)\n df = _get_future_daily_history(underlying_symbol, trade_date, window)\n if df is None or df.empty:\n return None, None, None, None, None\n\n highs = df[\"high\"].values\n lows = df[\"low\"].values\n closes = df[\"close\"].values\n opens = df[\"open\"].values\n\n if len(df) < max(N1, N2, atr_window):\n return None, None, None, None, None\n\n # 通道信号部分：使用最近N1/N2个交易日的数据\n high_N1 = highs[-N1:]\n low_N1 = lows[-N1:]\n high_N2 = highs[-N2:]\n low_N2 = lows[-N2:]\n\n upper = float(np.max(high_N1))\n lower = float(np.min(low_N1))\n exit_long = float(np.min(low_N2))\n exit_short = float(np.max(high_N2))\n\n # ATR计算：True Range = max(high-low, |high-prev_close|, |low-prev_close|)\n tr_list = []\n for i in range(len(df)):\n if i == 0:\n prev_close = opens[i]\n else:\n prev_close = closes[i - 1]\n high_i = highs[i]\n low_i = lows[i]\n tr = max(high_i - low_i,\n abs(high_i - prev_close),\n abs(low_i - prev_close))\n tr_list.append(tr)\n\n if len(tr_list) < atr_window:\n return None, None, None, None, None\n atr = float(np.mean(tr_list[-atr_window:]))\n\n return upper, lower, exit_long, exit_short, atr\n except Exception as e:\n print(f\"计算海龟通道/ATR失败: {e}\")\n return None, None, None, None, None\n\n\ndef handle_data(context, data):\n \"\"\"海龟交易法则主逻辑（黄金主力合约）。\"\"\"\n trade_date = context.now # yyyymmdd\n\n # 避免同一交易日重复下单\n if context.last_trade_date == trade_date:\n return\n\n # 1. 获取“昨日”的交易日，并以昨日的主力合约作为今日交易合约\n try:\n prev_trade_date = panda_data.get_previous_trading_date(\n date=trade_date,\n exchange=\"SH\",\n n=1\n )\n except Exception as e:\n print(f\"{trade_date}: 获取前一交易日失败: {e}\")\n return\n\n if not prev_trade_date:\n print(f\"{trade_date}: 无前一交易日信息，跳过\")\n return\n\n main_contract = _get_main_contract_symbol(prev_trade_date, context.underlying_symbol)\n if main_contract is None:\n print(f\"{trade_date}: 未能获取昨日({prev_trade_date})黄金主力合约，跳过\")\n return\n context.current_contract = main_contract\n\n # 2. 获取当日bar（仍然用今日的data和昨日主力合约）\n try:\n bar = data[main_contract]\n except Exception:\n print(f\"{trade_date}: data中无 {main_contract} 的bar，跳过\")\n return\n\n close_price = bar.close\n if close_price is None or close_price <= 0:\n print(f\"{trade_date}: 收盘价异常 {close_price}，跳过\")\n return\n\n # 3. 计算/获取当日海龟信号（基于品种主力历史，不依赖具体合约代码）\n if context.cached_signal[\"date\"] != trade_date:\n upper, lower, exit_long, exit_short, atr = _calc_turtle_signals(\n trade_date=trade_date,\n underlying_symbol=context.underlying_symbol,\n N1=context.N1,\n N2=context.N2,\n atr_window=context.atr_window,\n )\n if upper is None or lower is None or atr is None or atr <= 0:\n print(f\"{trade_date}: 信号数据不足或ATR异常，跳过\")\n return\n context.cached_signal = {\n \"date\": trade_date,\n \"upper\": upper,\n \"lower\": lower,\n \"exit_long\": exit_long,\n \"exit_short\": exit_short,\n \"atr\": atr,\n }\n else:\n upper = context.cached_signal[\"upper\"]\n lower = context.cached_signal[\"lower\"]\n exit_long = context.cached_signal[\"exit_long\"]\n exit_short = context.cached_signal[\"exit_short\"]\n atr = context.cached_signal[\"atr\"]\n\n print(f\"{trade_date} 使用昨日({prev_trade_date})主力 {main_contract} 收盘={close_price:.2f}, 上轨(N1)={upper:.2f}, 下轨(N1)={lower:.2f}, 退多(N2)={exit_long:.2f}, ATR={atr:.2f}\")\n\n # 4. 获取账户、持仓\n futures_account = context.future_account_dict.get(context.account)\n if not futures_account:\n print(f\"{trade_date}: 未找到期货账户 {context.account}\")\n return\n\n positions = futures_account.positions\n pos = positions.get(main_contract, None)\n long_qty = pos.buy_quantity if pos else 0\n short_qty = pos.sell_quantity if pos else 0\n\n # 5. 计算每个单位的手数（基于ATR）\n total_value = futures_account.total_value if futures_account.total_value else 0\n if total_value <= 0:\n print(f\"{trade_date}: 总权益异常 {total_value}, 跳过\")\n return\n\n # 合约乘数\n contract_mul = 1.0\n try:\n df_mul = panda_data.get_future_list(\n symbol=[main_contract],\n fields=[\"symbol\", \"contract_multiplier\"],\n is_trading=None\n )\n if df_mul is not None and not df_mul.empty:\n mul_val = df_mul.iloc[0].get(\"contract_multiplier\")\n if mul_val is not None and mul_val > 0:\n contract_mul = float(mul_val)\n except Exception as e:\n print(f\"获取合约乘数失败: {e}, 使用默认1\")\n\n # 每单位风险资金\n unit_risk_cash = total_value * context.account_risk\n # 每单位价格风险 ~ ATR\n per_lot_risk = atr * contract_mul\n if per_lot_risk <= 0:\n print(f\"{trade_date}: 每手风险为0，跳过\")\n return\n\n units_lots_float = unit_risk_cash / per_lot_risk\n unit_lots = int(units_lots_float)\n if unit_lots <= 0:\n unit_lots = 1\n\n # 根据当前持仓推断单位数\n if long_qty > 0:\n context.units_held = max(int(round(long_qty / max(unit_lots, 1))), 1)\n elif short_qty > 0:\n context.units_held = max(int(round(short_qty / max(unit_lots, 1))), 1)\n else:\n context.units_held = 0\n\n # 6. 退出信号优先处理\n traded = False\n if long_qty > 0:\n closable_long = pos.closable_buy_quantity if pos else long_qty\n if exit_long is not None and close_price < exit_long and closable_long > 0:\n print(f\"{trade_date}: 海龟退场信号，多头平仓 {main_contract} {closable_long} 手\")\n try:\n order = sell_close(context.account, main_contract, closable_long, style=MarketOrderStyle)[0]\n if order:\n print(f\"多头全部平仓成功, order_id={order.order_id}\")\n context.last_trade_date = trade_date\n context.units_held = 0\n context.entry_price = None\n context.last_pyramid_price = None\n traded = True\n except Exception as e:\n print(f\"多头平仓失败: {e}\")\n\n if context.allow_short and short_qty > 0 and not traded:\n closable_short = pos.closable_sell_quantity if pos else short_qty\n if exit_short is not None and close_price > exit_short and closable_short > 0:\n print(f\"{trade_date}: 空头退场信号，空头平仓 {main_contract} {closable_short} 手\")\n try:\n order = buy_close(context.account, main_contract, closable_short, style=MarketOrderStyle)[0]\n if order:\n print(f\"空头全部平仓成功, order_id={order.order_id}\")\n context.last_trade_date = trade_date\n context.units_held = 0\n context.entry_price = None\n context.last_pyramid_price = None\n traded = True\n except Exception as e:\n print(f\"空头平仓失败: {e}\")\n\n if traded:\n return\n\n # 7. 开仓/加仓逻辑（默认只做多）\n # 当前净持仓方向\n if long_qty == 0 and short_qty == 0:\n # 首次开多：突破N1最高价\n if close_price > upper:\n lots = unit_lots\n if lots <= 0:\n lots = 1\n print(f\"{trade_date}: 海龟入场信号，首次开多 {main_contract} {lots} 手\")\n try:\n order = buy_open(context.account, main_contract, lots, style=MarketOrderStyle)[0]\n if order:\n print(f\"首次开多成功, order_id={order.order_id}\")\n context.last_trade_date = trade_date\n context.units_held = 1\n context.entry_price = close_price\n context.last_pyramid_price = close_price\n except Exception as e:\n print(f\"首次开多失败: {e}\")\n # 如需做空，可加入：close_price < lower 时 sell_open\n elif context.allow_short and close_price < lower:\n lots = unit_lots\n if lots <= 0:\n lots = 1\n print(f\"{trade_date}: 海龟入场信号，首次开空 {main_contract} {lots} 手\")\n try:\n order = sell_open(context.account, main_contract, lots, style=MarketOrderStyle)[0]\n if order:\n print(f\"首次开空成功, order_id={order.order_id}\")\n context.last_trade_date = trade_date\n context.units_held = 1\n context.entry_price = close_price\n context.last_pyramid_price = close_price\n except Exception as e:\n print(f\"首次开空失败: {e}\")\n else:\n # 已有多头 => 分批加仓\n if long_qty > 0 and context.units_held < context.max_units:\n # 加仓条件: 价格在上一次加仓价基础上，每上涨 pyramid_step_atr * ATR 即加一单位\n trigger_price = context.last_pyramid_price + context.pyramid_step_atr * atr if context.last_pyramid_price is not None else None\n if trigger_price is not None and close_price > trigger_price:\n lots = unit_lots\n if lots <= 0:\n lots = 1\n print(f\"{trade_date}: 海龟加仓信号，在{trigger_price:.2f}之上加多 {main_contract} {lots} 手\")\n try:\n order = buy_open(context.account, main_contract, lots, style=MarketOrderStyle)[0]\n if order:\n print(f\"加仓成功, order_id={order.order_id}\")\n context.last_trade_date = trade_date\n context.units_held += 1\n context.last_pyramid_price = close_price\n except Exception as e:\n print(f\"加仓失败: {e}\")\n # 做空加仓逻辑（若开启做空）\n if context.allow_short and short_qty > 0 and context.units_held < context.max_units:\n trigger_price_short = context.last_pyramid_price - context.pyramid_step_atr * atr if context.last_pyramid_price is not None else None\n if trigger_price_short is not None and close_price < trigger_price_short:\n lots = unit_lots\n if lots <= 0:\n lots = 1\n print(f\"{trade_date}: 海龟空头加仓信号，在{trigger_price_short:.2f}之下加空 {main_contract} {lots} 手\")\n try:\n order = sell_open(context.account, main_contract, lots, style=MarketOrderStyle)[0]\n if order:\n print(f\"空头加仓成功, order_id={order.order_id}\")\n context.last_trade_date = trade_date\n context.units_held += 1\n context.last_pyramid_price = close_price\n except Exception as e:\n print(f\"空头加仓失败: {e}\")\n\n\ndef before_trading(context):\n futures_account = context.future_account_dict.get(context.account)\n if futures_account:\n print(f\"{context.now} 开盘前：总权益={futures_account.total_value}, 可用资金={futures_account.cash}\")\n\n\ndef after_trading(context):\n futures_account = context.future_account_dict.get(context.account)\n if futures_account:\n print(f\"{context.now} 收盘后：总权益={futures_account.total_value}, 持仓盈亏={futures_account.holding_pnl}\")\n for sym, pos in futures_account.positions.items():\n print(f\" {sym}: 多={pos.buy_quantity}, 空={pos.sell_quantity}, 总盈亏={pos.pnl}\")\n"
},
"model_path": "",
"output_db_id": null
},
{
"uuid": "507a7fde-99df-4fa2-9d51-3bbc288b628c",
"title": "期货回测",
"name": "FutureBacktestControl",
"type": "FutureBacktestControl",
"litegraph_id": 2,
"positionX": 913.8436889648438,
"positionY": 240.8333282470703,
"width": 210,
"height": 228,
"static_input_data": {
"code": "",
"start_future_capital": 10000000,
"commission_rate": 1,
"margin_rate": 1,
"frequency": "1d",
"start_date": "20250122",
"end_date": "20251231"
},
"model_path": "",
"output_db_id": null
},
{
"uuid": "f3bac0f9-e547-4ee5-aa76-cbc3fa460975",
"title": "策略回测结果",
"name": "BackTestResultControl",
"type": "BackTestResultControl",
"litegraph_id": 3,
"positionX": 1233.34375,
"positionY": 243.00001525878906,
"width": 210,
"height": 63,
"static_input_data": {
"task_id": "error"
},
"model_path": "",
"output_db_id": null
}
],
"links": [
{
"uuid": "19d15f01-dadb-41cb-a373-1e4fdb14701f",
"litegraph_id": 1,
"status": 1,
"previous_node_uuid": "7b1f3865-c076-4919-90b0-dbbab30f1fd5",
"next_node_uuid": "507a7fde-99df-4fa2-9d51-3bbc288b628c",
"output_field_name": "code",
"input_field_name": "code"
},
{
"uuid": "7d6b3e50-4cd7-4a4b-bc3f-b928d36a6fa8",
"litegraph_id": 2,
"status": 1,
"previous_node_uuid": "507a7fde-99df-4fa2-9d51-3bbc288b628c",
"next_node_uuid": "f3bac0f9-e547-4ee5-aa76-cbc3fa460975",
"output_field_name": "backtest_id",
"input_field_name": "task_id"
}
],
"id": "69884ae7fc307ce8e78facb2"
}
评论