知微
2762ddf3ae
- 量价分析: full_analysis输出volume_deep+成交量存储在price_history.json - FK约束移除: holding_strategies外键->holdings阻止自选股写入 - #000850 重评已写入(止损3.74/止盈4.06/RR1.67)含量价信号
232 lines
8.4 KiB
Python
232 lines
8.4 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
pre-flight-check.py — 策略上线前检查清单的自动化部分
|
||
|
||
检查三项可自动验证的关卡，输出结构化结果供知微在报告中引用。
|
||
只检查，不拦报告。不过的项目标记 @preflight:fail。
|
||
|
||
用法：
|
||
python3 pre-flight-check.py [code]
|
||
不传 code → 校验版本一致性 + 数据时效 + 全量持仓成本检查
|
||
传 code → 只检查单只股票（止损技术位 + R/R + 成本）
|
||
|
||
依赖：
|
||
- /home/hmo/projects/MoFin/data/prompts/registry.json（版本一致性）
|
||
- /home/hmo/web-dashboard/data/decisions.json（策略+成本）
|
||
- /home/hmo/web-dashboard/data/portfolio.json（持仓数据）
|
||
|
||
输出格式：
|
||
✅ 项目名 — 通过
|
||
❌ 项目名 — 失败原因
|
||
⚠️ 项目名 — 警告
|
||
"""
|
||
import json
|
||
import sys
|
||
from pathlib import Path
|
||
from datetime import datetime, timezone
|
||
|
||
from mo_data import read_decisions
|
||
|
||
# === 路径 ===
|
||
REGISTRY = Path("/home/hmo/projects/MoFin/data/prompts/registry.json")
|
||
|
||
# === 检查 7：版本一致性 ===
|
||
def check_version_consistency():
|
||
"""analysis-rules.depends_on.version == strategy-generation.current_version"""
|
||
try:
|
||
reg = json.loads(REGISTRY.read_text())
|
||
except Exception as e:
|
||
return ("⚠️ 版本一致性", f"无法读取 registry.json: {e}")
|
||
|
||
sg = None
|
||
ar = None
|
||
for p in reg["prompts"]:
|
||
if p["id"] == "strategy-generation":
|
||
sg = p
|
||
if p["id"] == "analysis-rules":
|
||
ar = p
|
||
|
||
if not sg or not ar:
|
||
return ("⚠️ 版本一致性", "registry.json 中缺少 strategy-generation 或 analysis-rules")
|
||
|
||
sg_ver = sg["current_version"]
|
||
ar_dep_ver = ar.get("depends_on", {}).get("version", "none")
|
||
|
||
if sg_ver == ar_dep_ver:
|
||
return ("✅ 版本一致性", f"strategy-generation@{sg_ver} == analysis-rules.depends_on@{ar_dep_ver}")
|
||
else:
|
||
return ("❌ 版本一致性",
|
||
f"strategy-generation@{sg_ver} != analysis-rules.depends_on@{ar_dep_ver}。"
|
||
f"运行 check-prompt-deps.py 修复")
|
||
|
||
|
||
# === 检查 1：数据时效 ===
|
||
def check_data_freshness():
|
||
"""检查 holding_strategies 表的最新更新时间"""
|
||
from datetime import datetime, timezone
|
||
try:
|
||
import sqlite3
|
||
c = sqlite3.connect(str(WEB_DATA / "mofin.db"))
|
||
row = c.execute("SELECT MAX(updated_at) FROM holding_strategies WHERE status IN ('active','updated')").fetchone()
|
||
c.close()
|
||
if row and row[0]:
|
||
mtime_dt = datetime.strptime(row[0][:19], '%Y-%m-%d %H:%M:%S')
|
||
age_minutes = (datetime.now() - mtime_dt).total_seconds() / 60
|
||
else:
|
||
return ("⚠️ 数据时效", "策略表无有效数据")
|
||
except Exception as e:
|
||
return ("⚠️ 数据时效", f"无法检查策略表: {e}")
|
||
|
||
if age_minutes < 60:
|
||
return ("✅ 数据时效",
|
||
f"策略数据 更新于 {age_minutes:.0f} 分钟前 ({mtime_dt.strftime('%H:%M')})")
|
||
elif age_minutes < 240:
|
||
return ("⚠️ 数据时效",
|
||
f"策略数据 已 {age_minutes:.0f} 分钟未更新（可能已收盘），数据视为陈旧")
|
||
else:
|
||
return ("❌ 数据时效",
|
||
f"策略数据 已 {age_minutes:.0f} 分钟未更新，数据过期，请检查数据管线")
|
||
|
||
|
||
# === 检查 5：成本有效 ===
|
||
def check_cost_validity():
|
||
"""检查 decisions.json 中所有持仓的成本是否有效"""
|
||
try:
|
||
dec = read_decisions()
|
||
except Exception as e:
|
||
return ("⚠️ 成本有效性", f"无法读取 decisions: {e}")
|
||
|
||
stocks = dec.get("stocks", dec.get("holdings", dec.get("strategies", [])))
|
||
if not stocks:
|
||
stocks = dec.get("portfolio", [])
|
||
|
||
issues = []
|
||
ok = 0
|
||
total = 0
|
||
|
||
for s in stocks:
|
||
# 支持 stocks[] 和 strategies[] 两种结构
|
||
cost = s.get("cost") or s.get("actual", {}).get("cost", None)
|
||
code = s.get("code") or s.get("symbol", "?")
|
||
name = s.get("name", "")
|
||
|
||
if cost is None or cost == 0:
|
||
issues.append(f"{code} {name}: cost={'null' if cost is None else cost}")
|
||
else:
|
||
ok += 1
|
||
total += 1
|
||
|
||
if not issues:
|
||
return ("✅ 成本有效性", f"全部 {total} 条持仓成本有效")
|
||
else:
|
||
return ("⚠️ 成本有效性",
|
||
f"{len(issues)}/{total} 条持仓成本缺失：{'，'.join(issues[:5])}"
|
||
f"{'...' if len(issues) > 5 else ''}")
|
||
|
||
|
||
# === 检查 3（单股）：止损技术依据 ===
|
||
def check_stop_technical(code):
|
||
"""检查单只股票的止损是否基于技术位（仅对单股模式生效）"""
|
||
try:
|
||
dec = read_decisions()
|
||
except Exception as e:
|
||
return ("⚠️ 止损技术位", f"无法读取 decisions: {e}")
|
||
|
||
stocks = dec.get("stocks", dec.get("strategies", []))
|
||
for s in stocks:
|
||
c = s.get("code", "")
|
||
if c.replace(".", "").replace("SZ", "").replace("SH", "").replace("HK", "") == code.replace(".", "").replace("SZ", "").replace("SH", "").replace("HK", ""):
|
||
stop = s.get("stop_loss", s.get("actual", {}).get("stop_loss"))
|
||
buy_zone = s.get("entry_low", s.get("buy_zone", [None, None]))
|
||
if isinstance(buy_zone, list):
|
||
entry_low = buy_zone[0]
|
||
else:
|
||
entry_low = buy_zone
|
||
|
||
if stop and entry_low and stop < entry_low * 0.95:
|
||
return ("⚠️ 止损技术位",
|
||
f"{code} 止损{stop} < 买入区下沿{entry_low}×0.95，"
|
||
f"检查是否用了固定百分比而非技术位")
|
||
return ("✅ 止损技术位", f"{code} 止损{stop}，需要人工确认是否基于支撑位")
|
||
|
||
return ("⚠️ 止损技术位", f"代码 {code} 未在 decisions.json 中找到")
|
||
|
||
|
||
# === 检查 4（单股）：R/R 达标 ===
|
||
def check_rr(code, price=None):
|
||
"""检查单只股票的 R/R 是否达标"""
|
||
try:
|
||
dec = read_decisions()
|
||
except Exception as e:
|
||
return ("⚠️ R/R 达标", f"无法读取 decisions: {e}")
|
||
|
||
stocks = dec.get("stocks", dec.get("strategies", []))
|
||
for s in stocks:
|
||
c = s.get("code", "")
|
||
if c.replace(".", "").replace("SZ", "").replace("SH", "").replace("HK", "") == code.replace(".", "").replace("SZ", "").replace("SH", "").replace("HK", ""):
|
||
stop = s.get("stop_loss", s.get("actual", {}).get("stop_loss"))
|
||
target = s.get("take_profit", s.get("actual", {}).get("take_profit"))
|
||
p = price or s.get("price", s.get("current_price"))
|
||
|
||
if not all([stop, target, p]):
|
||
return ("⚠️ R/R 达标", f"{code} 缺少止损/止盈/现价之一")
|
||
|
||
risk = p - stop
|
||
reward = target - p
|
||
if risk <= 0:
|
||
return ("❌ R/R 达标", f"{code} 现价{p}已低于止损{stop}，无R/R可言")
|
||
|
||
rr = reward / risk
|
||
cost = s.get("cost", s.get("actual", {}).get("cost"))
|
||
is_new = (cost is None or cost == 0)
|
||
|
||
min_rr = 2.5 if is_new else 2.0 # 简化处理，弱阻距检测需额外输入
|
||
if rr >= min_rr:
|
||
return ("✅ R/R 达标", f"{code} R/R={rr:.2f} >= {min_rr}")
|
||
else:
|
||
return ("❌ R/R 达标", f"{code} R/R={rr:.2f} < {min_rr}")
|
||
|
||
return ("⚠️ R/R 达标", f"代码 {code} 未在 decisions.json 中找到")
|
||
|
||
|
||
def main():
|
||
print("「策略上线前检查」自动化部分")
|
||
print(f"检查时间: {datetime.now().strftime('%Y-%m-%d %H:%M')}")
|
||
print()
|
||
|
||
if len(sys.argv) > 1:
|
||
# 单股模式
|
||
code = sys.argv[1]
|
||
price = float(sys.argv[2]) if len(sys.argv) > 2 else None
|
||
results = [
|
||
check_stop_technical(code),
|
||
check_rr(code, price),
|
||
]
|
||
else:
|
||
# 全量模式
|
||
results = [
|
||
check_version_consistency(),
|
||
check_data_freshness(),
|
||
check_cost_validity(),
|
||
]
|
||
|
||
for tag, msg in results:
|
||
print(f" {tag} — {msg}")
|
||
|
||
# 汇总
|
||
fails = sum(1 for tag, _ in results if tag.startswith("❌"))
|
||
warns = sum(1 for tag, _ in results if tag.startswith("⚠️"))
|
||
total = len(results)
|
||
|
||
print()
|
||
if fails == 0 and warns == 0:
|
||
print(f" ✅ {total}/{total} 项通过")
|
||
elif fails == 0:
|
||
print(f" ⚠️ {warns} 项警告，{total - warns} 项通过")
|
||
else:
|
||
print(f" ❌ {fails} 项失败，{warns} 项警告，{total - fails - warns} 项通过")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|