feat: 自适应策略系统 + 5只股票完整测试结论
- stock_adaptive.py: ADX环境感知策略切换(MA20/MACD) - 5只股票批量测试结果: - 茅台+五粮液(下跌): α=+1.1%/+39.2% ✅ 策略有效 - 平安银行(震荡): α=-3.8% - 宁德/沪深300(强势): α=-28.9%/-9.8% ❌ 策略有害 - 核心结论: 趋势明确时少动，趋势混乱时用MA20突破 - cangjie-skills INDEX.md更新测试结论
This commit is contained in:
小唯 A06
parent
25c0ecfdf5
commit
cfc14d15d7
|
|
@ -4,7 +4,7 @@
|
|||
更新: 2026-07-12 策略模拟结论
|
||||
状态: Phase5 模拟验证进行中
|
||||
|
||||
## 策略模拟结论（2026-07-12）
|
||||
## 自适应策略模拟结论（2026-07-12）
|
||||
|
||||
贵州茅台(600519) 2023-01 ~ 2026-07，买入持有基准: -13.8%
|
||||
|
||||
|
|
@ -17,6 +17,26 @@
|
|||
**结论**: MA20日突破策略最优，减少亏损+跑赢大盘
|
||||
**注意**: 贵州茅台整体下跌，所有策略均亏损，策略价值在于减少损失
|
||||
|
||||
---
|
||||
|
||||
## 自适应策略测试结果（2026-07-12）
|
||||
|
||||
ADX>25用MA20突破，ADX≤25用MACD。测5只股票:
|
||||
|
||||
| 股票 | 基准收益 | 自适应α | 评价 |
|
||||
|------|---------|---------|------|
|
||||
| 贵州茅台 | -13.8% | +1.1% ✅ | 下跌市有效 |
|
||||
| 五粮液 | -39.2% | +39.2% ✅ | 大幅跑赢 |
|
||||
| 平安银行 | +16.8% | -3.8% | 震荡市略输 |
|
||||
| 宁德时代 | +98.9% | -28.9% ❌ | 强势股策略干扰 |
|
||||
| 沪深300ETF | +36.2% | -9.8% ❌ | 强势ETF策略干扰 |
|
||||
| **平均** | - | **-0.5%** | 基本持平 |
|
||||
|
||||
**关键规律**:
|
||||
- 策略在**下跌趋势**中有效（茅台、五粮液）
|
||||
- 策略在**强势趋势**中有害（宁德+99%→策略只赚了70%）
|
||||
- **通用结论**: 趋势明确时少动，趋势混乱时用策略
|
||||
|
||||
## 已蒸馏 Skill
|
||||
|
||||
| # | 文件 | 主题 | 核心要点 |
|
||||
|
|
@ -33,10 +53,11 @@
|
|||
|
||||
Phase 5: 模拟交易与策略验证
|
||||
- ✅ 数据源: 腾讯/ifzq K线API (前复权日K)
|
||||
- ✅ 回测引擎: 纯Python (stock_compare.py)
|
||||
- ✅ 策略对比: MACD / MA突破 / 双均线
|
||||
- ⬜ 选股范围: 扩大多只股票测试
|
||||
- ⬜ 实盘模拟: 达到预期后小仓位实操
|
||||
- ✅ 回测引擎: 纯Python (stock_compare.py / stock_adaptive.py)
|
||||
- ✅ 策略对比: MACD / MA突破 / 双均线 / 自适应
|
||||
- ✅ 5只股票测试完成（茅台/五粮液/平安银行/宁德时代/沪深300ETF）
|
||||
- ⬜ 选股策略: 选弱势/震荡股，策略有效
|
||||
- ⬜ 小仓位实盘模拟: 选1-2只股票跑策略，验证真实效果
|
||||
|
||||
## 方法论来源
|
||||
|
||||
|
|
|
|||
|
|
@ -0,0 +1,341 @@
|
|||
#!/usr/bin/env python3
|
||||
"""
|
||||
小唯自适应策略 — 智能切换MACD/MA20/买入持有
|
||||
==========================================
|
||||
根据市场环境自动选择最优策略
|
||||
|
||||
判断逻辑:
|
||||
- 趋势强度(ADX) > 25 → 趋势市场 → MA20突破(追涨)
|
||||
- ADX <= 25 → 震荡市场 → MACD(区间波动)
|
||||
- 持仓时ADX突然下降 → 快速离场
|
||||
|
||||
用法:
|
||||
python3 stock_adaptive.py <代码> [起始] [结束]
|
||||
python3 stock_adaptive.py --batch # 批量测5只股票
|
||||
"""
|
||||
|
||||
import json, sys, urllib.request
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
|
||||
import numpy as np
|
||||
import pandas as pd
|
||||
|
||||
OUTPUT = Path.home() / ".hermes" / "stock_backtest"
|
||||
OUTPUT.mkdir(exist_ok=True)
|
||||
FEISHU_WEBHOOK = "https://open.feishu.cn/open-apis/bot/v2/hook/446db983-e392-4d2c-bfb8-f9060e5df3ad"
|
||||
|
||||
def send_feishu(msg):
|
||||
payload = json.dumps({"msg_type": "text", "content": {"text": msg}}).encode()
|
||||
req = urllib.request.Request(FEISHU_WEBHOOK, data=payload,
|
||||
headers={"Content-Type": "application/json"})
|
||||
try:
|
||||
with urllib.request.urlopen(req, timeout=10):
|
||||
pass
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
def get_data(code, start, end):
|
||||
mc = f"sh{code}" if code.startswith("6") else f"sz{code}"
|
||||
url = (f"https://web.ifzq.gtimg.cn/appstock/app/fqkline/get"
|
||||
f"?_var=kline_dayqfq¶m={mc},day,{start},{end},500,qfq")
|
||||
try:
|
||||
text = urllib.request.urlopen(url, timeout=10).read().decode("utf-8")
|
||||
data = json.loads(text.replace("kline_dayqfq=", "", 1))
|
||||
qfq = (data.get("data", {}).get(mc, {}).get("qfqday") or
|
||||
data.get("data", {}).get(mc, {}).get("day") or [])
|
||||
rows = []
|
||||
for item in qfq:
|
||||
if len(item) < 6:
|
||||
continue
|
||||
try:
|
||||
rows.append({"date": item[0], "open": float(item[1]),
|
||||
"close": float(item[2]), "high": float(item[3]),
|
||||
"low": float(item[4]), "volume": float(item[5])})
|
||||
except (ValueError, IndexError):
|
||||
continue
|
||||
df = pd.DataFrame(rows)
|
||||
df["date"] = pd.to_datetime(df["date"])
|
||||
df.set_index("date", inplace=True)
|
||||
df.sort_index(inplace=True)
|
||||
return df
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def compute_adx(df, n=14):
|
||||
"""计算ADX指标(趋势强度)，>25表示趋势市场"""
|
||||
high = df["high"]
|
||||
low = df["low"]
|
||||
close = df["close"]
|
||||
|
||||
# +DM, -DM
|
||||
plus_dm = high.diff()
|
||||
minus_dm = -low.diff()
|
||||
plus_dm[plus_dm < 0] = 0
|
||||
minus_dm[minus_dm < 0] = 0
|
||||
|
||||
# True Range
|
||||
tr1 = high - low
|
||||
tr2 = abs(high - close.shift())
|
||||
tr3 = abs(low - close.shift())
|
||||
tr = pd.concat([tr1, tr2, tr3], axis=1).max(axis=1)
|
||||
|
||||
# ADX
|
||||
atr = tr.rolling(n).mean()
|
||||
plus_di = (plus_dm.rolling(n).mean() / atr * 100).fillna(0)
|
||||
minus_di = (minus_dm.rolling(n).mean() / atr * 100).fillna(0)
|
||||
|
||||
dx = (abs(plus_di - minus_di) / (plus_di + minus_di + 1e-9) * 100).fillna(0)
|
||||
adx = dx.rolling(n).mean()
|
||||
return adx
|
||||
|
||||
|
||||
def adaptive_backtest(df, cash=100000):
|
||||
"""自适应策略: ADX>25用MA20，ADX<=25用MACD"""
|
||||
closes = df["close"].values
|
||||
n = len(closes)
|
||||
|
||||
# 预计算所有指标
|
||||
ma20 = pd.Series(closes).rolling(20).mean()
|
||||
adx = compute_adx(df, 14)
|
||||
adx_vals = adx.fillna(0).values
|
||||
|
||||
ema12 = pd.Series(closes).ewm(span=12).mean()
|
||||
ema26 = pd.Series(closes).ewm(span=26).mean()
|
||||
macd = ema12 - ema26
|
||||
signal = macd.ewm(span=9).mean()
|
||||
macd_vals = macd.values
|
||||
signal_vals = signal.values
|
||||
|
||||
# 持仓状态
|
||||
shares = 0
|
||||
c = cash
|
||||
peak = cash
|
||||
max_dd = 0.0
|
||||
trades = []
|
||||
regime_changes = [] # 记录策略切换
|
||||
|
||||
for i in range(26, n):
|
||||
price = closes[i]
|
||||
regime = "TREND" if adx_vals[i] > 25 else "RANGE"
|
||||
|
||||
# === MA20突破信号(趋势市场) ===
|
||||
if regime == "TREND":
|
||||
if shares == 0:
|
||||
if closes[i] > ma20.iloc[i] and closes[i-1] <= ma20.iloc[i-1]:
|
||||
s = int(c / price)
|
||||
if s > 0:
|
||||
c -= s * price
|
||||
shares = s
|
||||
trades.append(("BUY", df.index[i], s, price, "MA20_TREND"))
|
||||
elif shares > 0:
|
||||
if closes[i] < ma20.iloc[i] and closes[i-1] >= ma20.iloc[i-1]:
|
||||
c += shares * price
|
||||
trades.append(("SELL", df.index[i], shares, price, "MA20_TREND"))
|
||||
shares = 0
|
||||
|
||||
# === MACD信号(震荡市场) ===
|
||||
else: # RANGE
|
||||
if shares == 0:
|
||||
if macd_vals[i] > signal_vals[i] and macd_vals[i-1] <= signal_vals[i-1]:
|
||||
s = int(c / price)
|
||||
if s > 0:
|
||||
c -= s * price
|
||||
shares = s
|
||||
trades.append(("BUY", df.index[i], s, price, "MACD_RANGE"))
|
||||
elif shares > 0:
|
||||
if macd_vals[i] < signal_vals[i] and macd_vals[i-1] >= signal_vals[i-1]:
|
||||
c += shares * price
|
||||
trades.append(("SELL", df.index[i], shares, price, "MACD_RANGE"))
|
||||
shares = 0
|
||||
|
||||
equity = c + shares * price
|
||||
peak = max(peak, equity)
|
||||
dd = (peak - equity) / peak * 100 if peak > 0 else 0
|
||||
max_dd = max(max_dd, dd)
|
||||
|
||||
final = c + shares * closes[-1]
|
||||
ret = (final - cash) / cash * 100
|
||||
bh = (closes[-1] - closes[0]) / closes[0] * 100
|
||||
|
||||
# 统计
|
||||
total_trades = len([t for t in trades if t[0] == "SELL"])
|
||||
ma20_trades = len([t for t in trades if t[0] == "SELL" and t[4] == "MA20_TREND"])
|
||||
macd_trades = len([t for t in trades if t[0] == "SELL" and t[4] == "MACD_RANGE"])
|
||||
wins = len([t for t in trades if t[0] == "SELL" and t[3] > 0])
|
||||
|
||||
return {
|
||||
"final": final, "return": ret, "buyhold": bh,
|
||||
"alpha": ret - bh, "max_dd": max_dd,
|
||||
"total_trades": total_trades,
|
||||
"ma20_trades": ma20_trades,
|
||||
"macd_trades": macd_trades,
|
||||
"winrate": wins / total_trades * 100 if total_trades > 0 else 0,
|
||||
"trades": [(t[0], str(t[1].date()), t[2], t[3], t[4]) for t in trades],
|
||||
}
|
||||
|
||||
|
||||
def run_benchmark(df, cash=100000):
|
||||
"""跑三个基准策略用于对比"""
|
||||
closes = df["close"].values
|
||||
|
||||
# MACD
|
||||
ema12 = pd.Series(closes).ewm(span=12).mean()
|
||||
ema26 = pd.Series(closes).ewm(span=26).mean()
|
||||
macd = ema12 - ema26
|
||||
signal = macd.ewm(span=9).mean()
|
||||
shares = 0; c = cash; trades = 0
|
||||
for i in range(26, len(closes)):
|
||||
p = closes[i]
|
||||
if macd.iloc[i] > signal.iloc[i] and macd.iloc[i-1] <= signal.iloc[i-1]:
|
||||
if shares == 0:
|
||||
n = int(c / p); c -= n * p; shares = n
|
||||
elif macd.iloc[i] < signal.iloc[i] and macd.iloc[i-1] >= signal.iloc[i-1]:
|
||||
if shares > 0:
|
||||
c += shares * p; shares = 0; trades += 1
|
||||
final = c + shares * closes[-1]
|
||||
macd_ret = (final - cash) / cash * 100
|
||||
|
||||
# MA20
|
||||
ma20 = pd.Series(closes).rolling(20).mean()
|
||||
shares = 0; c = cash; trades = 0
|
||||
for i in range(20, len(closes)):
|
||||
p = closes[i]
|
||||
if closes[i] > ma20.iloc[i] and closes[i-1] <= ma20.iloc[i-1]:
|
||||
if shares == 0:
|
||||
n = int(c / p); c -= n * p; shares = n
|
||||
elif closes[i] < ma20.iloc[i] and closes[i-1] >= ma20.iloc[i-1]:
|
||||
if shares > 0:
|
||||
c += shares * p; shares = 0; trades += 1
|
||||
final = c + shares * closes[-1]
|
||||
ma20_ret = (final - cash) / cash * 100
|
||||
|
||||
bh = (closes[-1] - closes[0]) / closes[0] * 100
|
||||
return {"macd": macd_ret, "ma20": ma20_ret, "buyhold": bh}
|
||||
|
||||
|
||||
def analyze(code, start, end=None):
|
||||
end = end or datetime.now().strftime("%Y-%m-%d")
|
||||
df = get_data(code, start, end)
|
||||
if df is None or len(df) < 60:
|
||||
print(f"数据获取失败"); return
|
||||
|
||||
result = adaptive_backtest(df)
|
||||
bench = run_benchmark(df)
|
||||
|
||||
print(f"\n{'='*60}")
|
||||
print(f"自适应策略 vs 基准 | {code} | {start} ~ {end}")
|
||||
print(f"{'='*60}")
|
||||
print(f"{'策略':<22} {'收益':>10} {'α':>10} {'最大回撤':>10} {'交易':>6}")
|
||||
print(f"{'-'*60}")
|
||||
print(f"{'自适应(ADX切换)':<22} {result['return']:>+9.1f}% {result['alpha']:>+9.1f}% {result['max_dd']:>9.1f}% {result['total_trades']:>6}")
|
||||
print(f"{'MA20突破(固定)':<22} {bench['ma20']:>+9.1f}% {bench['ma20']-bench['buyhold']:>+9.1f}%")
|
||||
print(f"{'MACD(固定)':<22} {bench['macd']:>+9.1f}% {bench['macd']-bench['buyhold']:>+9.1f}%")
|
||||
print(f"{'买入持有':<22} {bench['buyhold']:>+9.1f}% {'基准':>9}")
|
||||
print(f"{'='*60}")
|
||||
print(f"自适应内: MA20触发{result['ma20_trades']}次, MACD触发{result['macd_trades']}次")
|
||||
print(f"胜率: {result['winrate']:.0f}%")
|
||||
|
||||
# 飞书
|
||||
msg = f"""📊 自适应策略对比
|
||||
|
||||
代码: {code} | {start} ~ {end} | {len(df)}条
|
||||
|
||||
策略收益对比:
|
||||
自适应(ADX切换): {result['return']:+.1f}% (α={result['alpha']:+.1f}%, 回撤{result['max_dd']:.1f}%, {result['total_trades']}笔)
|
||||
MA20突破(固定): {bench['ma20']:+.1f}%
|
||||
MACD(固定): {bench['macd']:+.1f}%
|
||||
买入持有基准: {bench['buyhold']:+.1f}%
|
||||
|
||||
自适应内: MA20触发{result['ma20_trades']}次, MACD触发{result['macd_trades']}次
|
||||
胜率: {result['winrate']:.0f}%
|
||||
|
||||
生成: {datetime.now().strftime('%Y-%m-%d %H:%M')}
|
||||
小唯股票投研 · 自适应策略"""
|
||||
send_feishu(msg)
|
||||
|
||||
result_file = OUTPUT / f"adaptive_{code}.json"
|
||||
with open(result_file, "w") as f:
|
||||
json.dump({"code": code, "start": start, "end": end,
|
||||
"adaptive": result, "benchmark": bench}, f, ensure_ascii=False, indent=2, default=str)
|
||||
print(f"数据存: {result_file}")
|
||||
return result
|
||||
|
||||
|
||||
def batch_test():
|
||||
"""批量测5只股票"""
|
||||
stocks = [
|
||||
("600519", "贵州茅台", "2023-01-01"),
|
||||
("000858", "五粮液", "2023-01-01"),
|
||||
("000001", "平安银行", "2023-01-01"),
|
||||
("300750", "宁德时代", "2023-01-01"),
|
||||
("510300", "沪深300ETF", "2023-01-01"),
|
||||
]
|
||||
end = datetime.now().strftime("%Y-%m-%d")
|
||||
results = []
|
||||
|
||||
print(f"\n{'='*70}")
|
||||
print(f"{'股票':<12} {'基准':>8} {'MA20':>8} {'MACD':>8} {'自适应α':>10} {'自适应回撤':>10} {'自适应胜率':>8}")
|
||||
print(f"{'-'*70}")
|
||||
|
||||
for code, name, start in stocks:
|
||||
df = get_data(code, start, end)
|
||||
if df is None:
|
||||
print(f"{name:<12} 数据获取失败")
|
||||
continue
|
||||
r = adaptive_backtest(df)
|
||||
b = run_benchmark(df)
|
||||
results.append({
|
||||
"name": name, "code": code,
|
||||
"buyhold": b["buyhold"],
|
||||
"ma20": b["ma20"],
|
||||
"macd": b["macd"],
|
||||
"adaptive": r["return"],
|
||||
"adaptive_alpha": r["alpha"],
|
||||
"adaptive_dd": r["max_dd"],
|
||||
"adaptive_wr": r["winrate"],
|
||||
"ma20_trades": r["ma20_trades"],
|
||||
"macd_trades": r["macd_trades"],
|
||||
})
|
||||
print(f"{name:<12} {b['buyhold']:>+7.1f}% {b['ma20']:>+7.1f}% {b['macd']:>+7.1f}% {r['alpha']:>+9.1f}% {r['max_dd']:>9.1f}% {r['winrate']:>7.0f}%")
|
||||
|
||||
print(f"{'='*70}")
|
||||
|
||||
# 汇总
|
||||
avg_alpha = np.mean([x["adaptive_alpha"] for x in results])
|
||||
win_count = sum(1 for x in results if x["adaptive_alpha"] > 0)
|
||||
print(f"\n自适应策略平均α: {avg_alpha:+.1f}% | 跑赢基准: {win_count}/{len(results)}只")
|
||||
|
||||
msg = f"""📊 自适应策略批量测试(5只)
|
||||
|
||||
| 股票 | 基准 | MA20 | MACD | 自适应α | 回撤 | 胜率 |
|
||||
|------|------|------|------|--------|------|------|
|
||||
"""
|
||||
for x in results:
|
||||
msg += f"| {x['name']} | {x['buyhold']:+.0f}% | {x['ma20']:+.0f}% | {x['macd']:+.0f}% | **{x['adaptive_alpha']:+.1f}%** | {x['adaptive_dd']:.0f}% | {x['adaptive_wr']:.0f}% |\n"
|
||||
|
||||
msg += f"""
|
||||
平均超额收益(α): **{avg_alpha:+.1f}%**
|
||||
跑赢基准: {win_count}/{len(results)}只
|
||||
|
||||
自适应策略: ADX>25用MA20突破, ADX≤25用MACD
|
||||
结论: {'有效 ✅' if avg_alpha > 0 else '无效 ❌'}
|
||||
|
||||
生成: {datetime.now().strftime('%Y-%m-%d %H:%M')}"""
|
||||
send_feishu(msg)
|
||||
print("\n✅ 批量测试完成，已推送飞书")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
if "--batch" in sys.argv:
|
||||
batch_test()
|
||||
elif len(sys.argv) < 2:
|
||||
print("用法: python3 stock_adaptive.py <代码> [起始] [结束]")
|
||||
print(" python3 stock_adaptive.py --batch")
|
||||
else:
|
||||
code = sys.argv[1]
|
||||
start = sys.argv[2] if len(sys.argv) > 2 else "2023-01-01"
|
||||
end = sys.argv[3] if len(sys.argv) > 3 else datetime.now().strftime("%Y-%m-%d")
|
||||
analyze(code, start, end)
|
||||
Loading…
Reference in New Issue