155 lines
5.5 KiB
Python
155 lines
5.5 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
小唯股票交易链路端到端演练
|
||
==========================
|
||
模拟完整生命周期：金叉信号 → 动态仓位开仓 → 持仓 → 死叉平仓 → 账户验证
|
||
|
||
验证项：
|
||
1. 信号生成（compute_signal → BUY/SELL）
|
||
2. 动态仓位分配（get_dynamic_alloc）
|
||
3. 多账户开仓（execute_signal BUY）
|
||
4. 账户状态更新（资金/持仓）
|
||
5. 平仓（execute_signal SELL → 盈亏计算）
|
||
6. 报表生成（report）
|
||
7. 清理（恢复演练前状态）
|
||
|
||
用法：python3 stock_trade_drill.py [--dry-run]
|
||
"""
|
||
import json, sys, os, shutil
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
|
||
sys.path.insert(0, str(Path(__file__).parent))
|
||
|
||
OUTPUT = Path.home() / ".hermes" / "stock_backtest"
|
||
DRILL_DIR = OUTPUT / "drill_backup"
|
||
MULTI_DIR = OUTPUT / "multi_account"
|
||
|
||
PASS = 0
|
||
FAIL = 0
|
||
|
||
|
||
def check(name, cond, detail=""):
|
||
global PASS, FAIL
|
||
if cond:
|
||
PASS += 1
|
||
print(f" ✅ {name} {detail}")
|
||
else:
|
||
FAIL += 1
|
||
print(f" ❌ {name} {detail}")
|
||
|
||
|
||
def backup():
|
||
"""备份多账户状态（演练后恢复）"""
|
||
if MULTI_DIR.exists():
|
||
if DRILL_DIR.exists():
|
||
shutil.rmtree(DRILL_DIR)
|
||
shutil.copytree(MULTI_DIR, DRILL_DIR)
|
||
|
||
|
||
def restore():
|
||
"""恢复多账户状态"""
|
||
if DRILL_DIR.exists():
|
||
if MULTI_DIR.exists():
|
||
shutil.rmtree(MULTI_DIR)
|
||
shutil.copytree(DRILL_DIR, MULTI_DIR)
|
||
shutil.rmtree(DRILL_DIR)
|
||
|
||
|
||
def main():
|
||
dry_run = "--dry-run" in sys.argv
|
||
print("=" * 60)
|
||
print(f"小唯股票交易链路演练 {datetime.now().strftime('%Y-%m-%d %H:%M')}")
|
||
print(f"模式: {'DRY-RUN（不落盘）' if dry_run else '真实演练（备份恢复）'}")
|
||
print("=" * 60)
|
||
|
||
if not dry_run:
|
||
backup()
|
||
|
||
import stock_multi_account as sma
|
||
import stock_portfolio
|
||
|
||
# ========== 1. 信号生成 ==========
|
||
print("\n【1. 信号生成】")
|
||
sig_file = OUTPUT / "ma20_result_000858.json"
|
||
check("回测结果存在", sig_file.exists())
|
||
if sig_file.exists():
|
||
sig_data = json.load(open(sig_file))
|
||
print(f" 五粮液回测: 策略收益{sig_data.get('strategy_return',0):+.1f}% α{sig_data.get('alpha',0):+.1f}%")
|
||
|
||
# 模拟真实金叉信号（用新能源 - 最强标的）
|
||
print("\n【2. 信号→开仓（新能源 宁德时代）】")
|
||
price = 395.30
|
||
# 2a. 动态仓位
|
||
alloc = sma.get_dynamic_alloc("新能源")
|
||
check("动态仓位=100%", alloc == 1.0, f"({alloc:.0%})")
|
||
|
||
# 2b. 开仓
|
||
action, detail = sma.execute_signal(
|
||
{"industry": "新能源", "signal": "BUY", "close": price}, dry_run=dry_run)
|
||
check("新能源开仓成功", action == "BUY", f"[{action}] {detail}")
|
||
if not dry_run:
|
||
acct = sma.load_account("新能源")
|
||
has_pos = bool(acct["positions"])
|
||
shares = sum(p["shares"] for p in acct["positions"]) if has_pos else 0
|
||
check("账户已持仓", has_pos, f"({shares}股)")
|
||
check("资金扣减", acct["current_capital"] < 100000,
|
||
f"(剩余{acct['current_capital']:.0f})")
|
||
|
||
# ========== 2. 弱势行业半仓 ==========
|
||
print("\n【3. 弱势行业动态仓位（白酒 五粮液）】")
|
||
alloc_bj = sma.get_dynamic_alloc("白酒")
|
||
check("白酒仓位<100%", alloc_bj < 1.0, f"({alloc_bj:.0%})")
|
||
action2, detail2 = sma.execute_signal(
|
||
{"industry": "白酒", "signal": "BUY", "close": 78.0}, dry_run=dry_run)
|
||
check("白酒开仓成功", action2 == "BUY", f"[{action2}] {detail2}")
|
||
if not dry_run:
|
||
acct2 = sma.load_account("白酒")
|
||
shares2 = sum(p["shares"] for p in acct2["positions"]) if acct2["positions"] else 0
|
||
cost2 = sum(p["shares"] * p["avg_cost"] for p in acct2["positions"]) if acct2["positions"] else 0
|
||
check("白酒半仓（约50%）", 0.3 < cost2 / 100000 < 0.7,
|
||
f"(成本{cost2:.0f} = {cost2/100000:.0%})")
|
||
|
||
# ========== 3. 平仓 ==========
|
||
print("\n【4. 平仓（新能源 死叉）】")
|
||
sell_price = price * 1.05 # 模拟盈利5%
|
||
action3, detail3 = sma.execute_signal(
|
||
{"industry": "新能源", "signal": "SELL", "close": sell_price}, dry_run=dry_run)
|
||
if dry_run:
|
||
# dry-run 不落盘，SELL 会因为账户空仓返回 HOLD —— 跳过状态断言
|
||
check("新能源平仓调用成功", action3 in ("SELL", "HOLD"),
|
||
f"[{action3}] {detail3}")
|
||
else:
|
||
check("新能源平仓成功", action3 == "SELL", f"[{action3}] {detail3}")
|
||
acct3 = sma.load_account("新能源")
|
||
check("平仓后空仓", not acct3["positions"])
|
||
check("有平仓记录", len(acct3["closed_trades"]) == 1)
|
||
trade = acct3["closed_trades"][0]
|
||
check("盈利计算", trade["pnl"] > 0, f"(盈亏{trade['pnl']:.0f})")
|
||
|
||
# ========== 4. 报表 ==========
|
||
print("\n【5. 账户报表】")
|
||
report = sma.cmd_report()
|
||
check("日报生成", "总资产" in report, f"({report.splitlines()[-1]})")
|
||
|
||
# ========== 5. 组合扫描联动 ==========
|
||
print("\n【6. 组合扫描联动】")
|
||
signals = stock_portfolio.build_portfolio_report(push=False)
|
||
check("组合扫描运行", isinstance(signals, list))
|
||
print(f" (今日金叉: {len(signals)} 只)")
|
||
|
||
# ========== 结果 ==========
|
||
print("\n" + "=" * 60)
|
||
print(f"演练结果: ✅ {PASS} 通过 / ❌ {FAIL} 失败")
|
||
print("=" * 60)
|
||
|
||
if not dry_run:
|
||
restore()
|
||
print("已恢复演练前账户状态")
|
||
|
||
return 0 if FAIL == 0 else 1
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|