186 lines
7.3 KiB
Python
186 lines
7.3 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
贵州茅台 MA20 突破策略回测
|
||
策略：金叉买入，死叉卖出
|
||
"""
|
||
|
||
import sys, json, urllib.request
|
||
from datetime import datetime, timedelta
|
||
from pathlib import Path
|
||
import numpy as np
|
||
import pandas as pd
|
||
|
||
OUTPUT = Path.home() / ".hermes" / "stock_backtest"
|
||
FEISHU_WEBHOOK = "https://open.feishu.cn/open-apis/bot/v2/hook/446db983-e392-4d2c-bfb8-f9060e5df3ad"
|
||
|
||
def get_data(code, count=500):
|
||
mc = f"sh{code}" if code.startswith("6") else f"sz{code}"
|
||
end = datetime.now().strftime("%Y-%m-%d")
|
||
start = (datetime.now() - timedelta(days=count*1.5)).strftime("%Y-%m-%d")
|
||
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
|
||
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])})
|
||
df = pd.DataFrame(rows)
|
||
df["date"] = pd.to_datetime(df["date"])
|
||
return df.sort_values("date").reset_index(drop=True)
|
||
except Exception as e:
|
||
print(f"数据获取失败: {e}")
|
||
return pd.DataFrame()
|
||
|
||
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 backtest_ma20(code, name, initial_cash=100000):
|
||
df = get_data(code, 500)
|
||
if df.empty:
|
||
print(f"获取数据失败")
|
||
return
|
||
|
||
df["ma20"] = df["close"].rolling(window=20).mean()
|
||
df["prev_close"] = df["close"].shift(1)
|
||
df["prev_ma20"] = df["ma20"].shift(1)
|
||
|
||
# 金叉/死叉信号
|
||
df["golden_cross"] = (df["prev_close"] < df["prev_ma20"]) & (df["close"] > df["ma20"])
|
||
df["death_cross"] = (df["prev_close"] > df["prev_ma20"]) & (df["close"] < df["ma20"])
|
||
|
||
# 策略回测
|
||
cash = initial_cash
|
||
position = 0
|
||
trades = []
|
||
equity = []
|
||
in_position = False
|
||
buy_price = 0
|
||
|
||
for i, row in df.iterrows():
|
||
if pd.isna(row["ma20"]):
|
||
equity.append({"date": row["date"], "value": cash})
|
||
continue
|
||
|
||
date_str = row["date"].strftime("%Y-%m-%d")
|
||
price = row["close"]
|
||
|
||
# 买入（金叉）
|
||
if row["golden_cross"] and not in_position:
|
||
shares = int(cash / price)
|
||
cost = shares * price
|
||
if shares > 0:
|
||
cash -= cost
|
||
position = shares
|
||
buy_price = price
|
||
in_position = True
|
||
trades.append({"type": "BUY", "date": date_str, "price": price, "shares": shares, "reason": "金叉"})
|
||
|
||
# 卖出（死叉）
|
||
elif row["death_cross"] and in_position:
|
||
proceeds = position * price
|
||
cash += proceeds
|
||
profit_pct = (price - buy_price) / buy_price * 100
|
||
trades.append({"type": "SELL", "date": date_str, "price": price, "shares": position, "profit_pct": profit_pct, "reason": "死叉"})
|
||
position = 0
|
||
in_position = False
|
||
buy_price = 0
|
||
|
||
# 当日价值
|
||
value = cash + position * price
|
||
equity.append({"date": date_str, "value": value, "price": price})
|
||
|
||
# 最终资产
|
||
final_price = df.iloc[-1]["close"]
|
||
final_value = cash + position * final_price
|
||
|
||
# 买入持有对比
|
||
buy_price_hold = df.iloc[19]["close"] # 第一个有效 MA20 时的价格
|
||
shares_hold = int(initial_cash / buy_price_hold)
|
||
hold_value = shares_hold * final_price
|
||
hold_return = (hold_value - initial_cash) / initial_cash * 100
|
||
strat_return = (final_value - initial_cash) / initial_cash * 100
|
||
alpha = strat_return - hold_return
|
||
|
||
# 最大回撤
|
||
equity_curve = [e["value"] for e in equity]
|
||
peak = equity_curve[0]
|
||
max_dd = 0
|
||
for v in equity_curve:
|
||
if v > peak: peak = v
|
||
dd = (peak - v) / peak * 100
|
||
if dd > max_dd: max_dd = dd
|
||
|
||
# 胜率
|
||
sell_trades = [t for t in trades if t["type"] == "SELL"]
|
||
win_trades = [t for t in sell_trades if t.get("profit_pct", 0) > 0]
|
||
win_rate = len(win_trades) / len(sell_trades) * 100 if sell_trades else 0
|
||
|
||
print(f"========================================")
|
||
print(f"{name}({code}) MA20突破策略回测")
|
||
print(f"数据区间: {df['date'].min().date()} ~ {df['date'].max().date()}")
|
||
print(f"========================================")
|
||
print(f" 初始资金: {initial_cash:,.0f}")
|
||
print(f" 最终资产: {final_value:,.0f}")
|
||
print(f" 策略收益: {strat_return:+.2f}%")
|
||
print(f" 买入持有: {hold_return:+.2f}%")
|
||
print(f" 超额收益(α):{alpha:+.2f}%")
|
||
print(f" 最大回撤: {max_dd:.2f}%")
|
||
print(f" 交易次数: {len(sell_trades)}")
|
||
print(f" 胜率: {win_rate:.1f}%")
|
||
print()
|
||
print(f" 金叉次数: {df['golden_cross'].sum()}")
|
||
print(f" 死叉次数: {df['death_cross'].sum()}")
|
||
print(f" 当前持仓: {'是' if in_position else '否'}")
|
||
if in_position:
|
||
print(f" 持仓成本: {buy_price:.2f}")
|
||
print(f" 当前盈亏: {(final_price - buy_price) / buy_price * 100:+.2f}%")
|
||
print()
|
||
print(f" 最近5笔交易:")
|
||
for t in trades[-5:]:
|
||
if t["type"] == "BUY":
|
||
print(f" [{t['date']}] BUY {t['shares']}股@{t['price']:.2f} ({t['reason']})")
|
||
else:
|
||
print(f" [{t['date']}] SELL {t['shares']}股@{t['price']:.2f} {t['profit_pct']:+.2f}% ({t['reason']})")
|
||
|
||
# 保存结果
|
||
result = {
|
||
"code": code, "name": name, "strategy": "MA20",
|
||
"period": f"{df['date'].min().date()} ~ {df['date'].max().date()}",
|
||
"initial_cash": initial_cash, "final_value": final_value,
|
||
"strategy_return": strat_return, "buyhold_return": hold_return,
|
||
"alpha": alpha, "max_drawdown": max_dd,
|
||
"total_trades": len(sell_trades), "win_rate": win_rate,
|
||
"trades": str(trades[-10:]),
|
||
"in_position": in_position, "buy_price": buy_price if in_position else None,
|
||
"current_price": float(final_price)
|
||
}
|
||
OUTPUT.mkdir(exist_ok=True)
|
||
result_file = OUTPUT / f"ma20_result_{code}.json"
|
||
with open(result_file, "w") as f:
|
||
json.dump(result, f, ensure_ascii=False, indent=2)
|
||
|
||
# 飞书推送
|
||
msg = (f"📊 {name}({code}) MA20策略回测\n"
|
||
f"区间: {df['date'].min().date()} ~ {df['date'].max().date()}\n"
|
||
f"策略收益: {strat_return:+.2f}% | 买入持有: {hold_return:+.2f}% | α={alpha:+.2f}%\n"
|
||
f"最大回撤: {max_dd:.2f}% | 交易次数: {len(sell_trades)} | 胜率: {win_rate:.1f}%\n"
|
||
f"当前: {'持仓中 ' + str(round((final_price-buy_price)/buy_price*100,2)) + '%' if in_position else '空仓'}")
|
||
send_feishu(msg)
|
||
|
||
if __name__ == "__main__":
|
||
if len(sys.argv) >= 3:
|
||
backtest_ma20(sys.argv[1], sys.argv[2], 100000)
|
||
else:
|
||
backtest_ma20("600519", "贵州茅台", 100000) |