116 lines
3.8 KiB
Python
116 lines
3.8 KiB
Python
#!/usr/bin/env python3
|
|
"""data_flow_audit.py — 数据流架构审计
|
|
|
|
对比实际读写关系 vs 设计意图（预期读写方），标记违规。
|
|
输出给 mofin_health.py 消费，在 Dashboard 数据流 Tab 显示。
|
|
"""
|
|
import re, os, json, subprocess
|
|
from pathlib import Path
|
|
|
|
PROFILE_SCRIPTS = Path("/home/hmo/.hermes/profiles/position-analyst/scripts")
|
|
MOFIN_SCRIPTS = Path("/home/hmo/MoFin/scripts")
|
|
SCRIPTS_DIR = PROFILE_SCRIPTS if PROFILE_SCRIPTS.exists() else MOFIN_SCRIPTS
|
|
|
|
# ── 设计意图注册表 ──
|
|
# 每张核心表的预期写入方（谁应该写）+ 预期用途
|
|
DESIGN_INTENT = {
|
|
"live_prices": {
|
|
"expected_writers": ["price_monitor"],
|
|
"policy": "single_writer",
|
|
"desc": "实时价格缓存 — 仅price_monitor写入，其他一律读",
|
|
"remarks": "违规: 自行拉API会导致并发写+重复请求"
|
|
},
|
|
"holdings": {
|
|
"expected_writers": ["import_holding_xls", "dad_asset_update"],
|
|
"policy": "restricted",
|
|
"desc": "持仓数据 — 仅通过成交截图导入或Dad确认更新",
|
|
"remarks": ""
|
|
},
|
|
"portfolio_summary": {
|
|
"expected_writers": ["import_holding_xls", "dad_asset_update", "price_monitor"],
|
|
"policy": "restricted",
|
|
"desc": "组合汇总 — 持仓导入+价格更新",
|
|
"remarks": ""
|
|
},
|
|
"cash_log": {
|
|
"expected_writers": ["import_holding_xls", "dad_asset_update"],
|
|
"policy": "restricted",
|
|
"desc": "资金流水 — 仅截图导入或Dad确认",
|
|
"remarks": ""
|
|
},
|
|
"holding_strategies": {
|
|
"expected_writers": ["strategy_review", "strategy_evaluator", "per_stock_reassess"],
|
|
"policy": "multi_writer",
|
|
"desc": "策略数据 — 多个分析流程可写",
|
|
"remarks": ""
|
|
},
|
|
"price_events": {
|
|
"expected_writers": ["price_monitor"],
|
|
"policy": "single_writer",
|
|
"desc": "价格触发事件 — 仅price_monitor写入",
|
|
"remarks": ""
|
|
},
|
|
"sector_snapshots": {
|
|
"expected_writers": ["market_watch"],
|
|
"policy": "single_writer",
|
|
"desc": "板块快照 — 仅market_watch写入",
|
|
"remarks": ""
|
|
},
|
|
}
|
|
|
|
# 直接拉API的价格违规扫描
|
|
API_PATTERNS = [
|
|
r"qt\.gtimg\.cn",
|
|
r"tencent.*quote",
|
|
r"get_quote",
|
|
r"fetch.*price",
|
|
r"stock_quote\.",
|
|
]
|
|
|
|
def scan_price_violations():
|
|
"""扫描不走live_prices直接拉API的脚本"""
|
|
violations = []
|
|
for py_file in sorted(SCRIPTS_DIR.glob("*.py")):
|
|
name = py_file.stem
|
|
if name in ("price_monitor", "stock_quote", "mofin_db", "mo_data", "deploy_sync"):
|
|
continue # 这些是基础设施/被允许的
|
|
content = py_file.read_text()
|
|
for pat in API_PATTERNS:
|
|
if re.search(pat, content):
|
|
# 找具体行号
|
|
lines = content.split("\n")
|
|
for i, line in enumerate(lines, 1):
|
|
if re.search(pat, line):
|
|
violations.append({
|
|
"script": name,
|
|
"line": i,
|
|
"code": line.strip()[:80],
|
|
})
|
|
break
|
|
return violations
|
|
|
|
|
|
def audit():
|
|
"""读取当前数据流扫描结果，叠加设计意图"""
|
|
from mo_data import get_flow
|
|
# 这里应该读mofin_health输出的entities
|
|
|
|
violations = scan_price_violations()
|
|
|
|
result = {
|
|
"design_intent": DESIGN_INTENT,
|
|
"price_api_violations": violations,
|
|
"summary": {
|
|
"total_violations": len(violations),
|
|
"violating_scripts": list(set(v["script"] for v in violations)),
|
|
}
|
|
}
|
|
|
|
return result
|
|
|
|
|
|
if __name__ == "__main__":
|
|
import json
|
|
result = audit()
|
|
print(json.dumps(result, ensure_ascii=False, indent=2))
|