用 AKQuant 测试翻倍股（doubles）策略
来源: Trilium
─────────────────────────────────────── ● 请考虑如何使用akquant来测试我的http://192.168.123.104:3001/john/doubles.git ────────────────────────────────────────
🔐 Nous agent key refreshed after 401. Retrying request...
D:\hermes\mystocks\doubles\test_akquant.py 02m231s)
╭─ ⚕ Hermes ────────────────────────────────────────── 让我先回顾 doubles 项目的结构和策略逻辑： ╰───────────────────────────────────────── ┊ 🔍 preparing session_search… ┊ 🔍 recall "doubles 翻倍股 Ross Hook 策略" 0.7s ┊ 🔍 preparing session_search… ┊ 🔍 recall "doubled_stocks_ma_analysis MA斜率 五阶段" 0.1s ┊ ⚡ preparing mempalace_search…
╭─ ⚕ Hermes ────────────────────────────────────────── 基于我的记忆和文档内容，给你一个完整的方案：
用 AKQuant 测试翻倍股（doubles）策略
你的 doubles 项目核心逻辑
你的研究已经得出几条关键结论：
| 结论 | 说明 |
|---|---|
| 起涨信号：69% 翻倍股起涨于 MA20 < MA60（死叉） | 短期均线下穿长期均线后，反而爆发反弹 |
| MA60 全程正斜率 | 长期趋势向上是翻倍的前提条件 |
| MA20 高斜率 +3.2%/天 | 见顶点信号，斜率过高则追高风险大 |
| 五阶段模型 | 起涨前→起涨日→起涨后10天→见顶前10天→见顶日 |
用 AKQuant 实现回测方案
AKQuant 可以直接把上述逻辑编码为策略，在一个 Framework 内完成 筛选 → 回测 → 绩效评估 的闭环。
第一步：准备数据
AKQuant 最适配的数据源是你的 ClickHouse day_kline 表（5207股、2024-01 至 2026-06）。用 python-clickhouse 或转换脚本把 ClickHouse 数据导出为 AKQuant 标准格式：
python import akshare as ak import pandas as pd from akquant import Strategy, run_backtest
方法1：直接通过 AKShare 取 A 股日线 df = ak.stock_zh_a_hist( symbol="000001.SH", # 或 "sz000001" period="daily", start_date="20240301", end_date="20260612", adjust="qfq", # 前复权 — 必须！ )
df = df.rename(columns={ "日期": "date", "开盘": "open", "最高": "high", "最低": "low", "收盘": "close", "成交量": "volume", }) df['date'] = pd.to_datetime(df['date']) df['symbol'] = "000001"
第二步：编码双均线策略（你的核心逻辑）
python import numpy as np from akquant import Strategy, run_backtest
class RossHookStrategy(Strategy): """基于 MA20/MA60 死叉起涨信号的翻倍股筛选策略
入场条件（基于你的271只翻倍股分析结论）:
- MA20 < MA60（死叉，69%翻倍股起涨于此）
- MA60 斜率 > 0（长期趋势向上，翻倍前提）
- MA20 斜率 < 0（短期回调正在加速）
出场条件:
- MA20 斜率 > +3.2%/天（见顶信号）
- 固定止损（如 -8%）
- 固定止盈（如 +100%）
"""
def init(self, ma_short=20, ma_long=60,
stop_loss=-0.08, take_profit=2.0,
ma_slope_peak=0.032):
self.ma_short = ma_short
self.ma_long = ma_long
self.stop_loss = stop_loss
self.take_profit = take_profit
self.ma_slope_peak = ma_slope_peak
self.warmup_period = ma_long + 5 # 预热期保证均线可用
def on_bar(self, bar):
hist = self.get_history(count=self.ma_long + 5, field="close")
if len(hist) < self.ma_long + 2:
return
closes = hist
ma20_now = closes[-20:].mean() if len(closes) >= 20 else None
ma60_now = closes[-60:].mean() if len(closes) >= 60 else None
ma20_prev = closes[-21:-1].mean() if len(closes) >= 21 else None
ma60_prev = closes[-61:-1].mean() if len(closes) >= 61 else None
if not all([ma20_now, ma60_now, ma20_prev, ma60_prev]):
return
position = self.get_position(bar.symbol)
# 计算 MA20 斜率（最近5根K线变化率）
ma20_old = closes[-25:-21].mean() if len(closes) >= 25 else ma20_prev
ma20_slope = (ma20_now - ma20_old) / ma20_old
# === 入场逻辑：死叉 + MA60上行 + MA20下行 ===
if position == 0:
if ma20_now < ma60_now and ma60_prev <= ma60_now and ma20_slope < 0:
self.buy(bar.symbol, 100)
self.log(f"[{bar.timestamp_iso}] 死叉买入 {bar.symbol} @ {bar.close:.2f} "
f"MA20={ma20_now:.2f} MA60={ma60_now:.2f} slope={ma20_slope*100:.3f}%")
# === 出场逻辑：MA20斜率过高 / 止损 / 止盈 ===
elif position > 0:
entry_price = self.get_instrument_field(bar.symbol, "entry_price") or bar.close / (1 + self.stop_loss)
pnl_pct = (bar.close - entry_price) / entry_price
# 斜率过高 = 见顶信号
if ma20_slope > self.ma_slope_peak:
self.sell(bar.symbol, 100)
self.log(f"[{bar.timestamp_iso}] 斜率见顶卖出 {bar.symbol} @ {bar.close:.2f} "
f"slope={ma20_slope100:.3f}% pnl={pnl_pct100:.1f}%")
# 止损
elif pnl_pct < self.stop_loss:
self.sell(bar.symbol, 100)
self.log(f"[{bar.timestamp_iso}] 止损卖出 {bar.symbol} @ {bar.close:.2f} "
f"pnl={pnl_pct*100:.1f}%")
# 止盈（翻倍股）
elif pnl_pct >= self.take_profit:
self.sell(bar.symbol, 100)
self.log(f"[{bar.timestamp_iso}] 止盈卖出 {bar.symbol} @ {bar.close:.2f} "
f"pnl={pnl_pct*100:.1f}%")
第三步：批量回测271只翻倍股
python 用你的 CSV 获取271只翻倍股代码 import pandas as pd candidates = pd.read_csv("D:/hermes/mystocks/doubles/doubled_stocks_list.csv", dtype=str)['代码'].tolist()
results = [] for code in candidates: try: # 整 AKShare 取该股前复权日线（2024-01 往后足够覆盖回测期） df = ak.stock_zh_a_hist( symbol=code, period="daily", start_date="20240101", end_date="20260612", adjust="qfq", ) if len(df) < 120: # 少于120天的跳过 continue df = df.rename(columns={"日期":"date","开盘":"open","最高":"high", "最低":"low","收盘":"close","成交量":"volume"}) df['date'] = pd.to_datetime(df['date']) df['symbol'] = code
result = run_backtest(
strategy=RossHookStrategy,
data=df,
symbols=code,
initial_cash=100000,
commission_rate=0.0003,
slippage=0.001,
show_progress=False,
)
m = result.metrics_df
results.append({
"code": code,
"total_return": m.loc["total_return_pct", "Value"],
"max_drawdown": m.loc["max_drawdown_pct", "Value"],
"sharpe": m.loc["sharpe_ratio", "Value"],
"trades": m.loc["trade_count", "Value"],
})
except Exception as e:
print(f" {code} 失败: {e}")
report = pd.DataFrame(results).sort_values("total_return", ascending=False) print(report) report.to_csv("D:/hermes/mystocks/doubles/akquant_backtest_results.csv", index=False)
第四步：参数扫描（优化入场/出场阈值）
python from akquant import run_grid_search
param_grid = { "ma_short": [10, 20], "ma_long": [40, 60], "stop_loss": [-0.05, -0.08, -0.10], "take_profit": [1.5, 2.0, 3.0], "ma_slope_peak": [0.02, 0.032, 0.05], }
results_grid = run_grid_search( strategy=RossHookStrategy, param_grid=param_grid, data=ddaily_df, # 可以先用一只股做快速参数扫描 initial_cash=100000, sort_by="sharpe_ratio", max_workers=4, )
print(results_grid.head(10))
对比你的原始数据 vs AKQuant 回测结果
你的原始数据库： - 271 只确认翻倍股（ClickHouse 筛选结果） - 265 只 MA 斜率分析（ELTDX K线 + 五阶段计算）
AKQuant 回测的价值在于： 1. 验证你的策略逻辑的真实收益 — 不用事后确认翻倍，看信号发出时是否真的买到了 2. 评估入场/出场规则的改进空间 — 正斜率、平斜率、反斜率各方案对收益的影响 3. 参数敏感性分析 — 20天 vs 30天均线？斜率阈值 3.2% 是否最优？ 4. 对比不同止损/止盈策略
实际执行路径
我建议分三步：
- 先跑一只股的快速验证：用 000001（平安银行）或 605319（无锡振华，你的A级候选），验证 RossHookStrategy 能正常产生信号和结果
- 批量回测265只：用你有 K 线数据的 265 只股跑策略，对比原始翻倍股的实际表现
- 参数优化：在最有潜力的子集上跑 grid_search
需要我现在帮你开始第1步——在 000001 上跑一次 RossHookStrategy 快速验证吗？ ────────────────────────────────────────── ● 同意