refactor: remove duplicate scripts + JSON constants + update docs
This commit is contained in:
知微
+88
-90
@@ -1,90 +1,88 @@
|
||||
#!/usr/bin/env python3
|
||||
"""data_validate.py — 数据自检，在所有报告产出前执行
|
||||
|
||||
检查清单：
|
||||
1. 总资产 = 市值 + 现金 （误差 < 1%）
|
||||
2. 持仓 vs 决策交叉检查
|
||||
3. 币种一致性（港股必须currency=HKD）
|
||||
4. 数据时效：portfolio.json/decisions.json 今日已更新
|
||||
|
||||
返回值：通过→退出码0，输出"OK"。失败→退出码1，输出问题描述。
|
||||
"""
|
||||
|
||||
import json, sys
|
||||
from datetime import datetime, timezone
|
||||
from mo_data import read_portfolio, read_decisions, read_watchlist
|
||||
|
||||
DATA_DIR = "/home/hmo/web-dashboard/data"
|
||||
PORTFOLIO_PATH = f"{DATA_DIR}/portfolio.json"
|
||||
DECISIONS_PATH = f"{DATA_DIR}/decisions.json"
|
||||
STALE_REPORT = f"{DATA_DIR}/strategy_staleness_report.json"
|
||||
|
||||
issues = []
|
||||
|
||||
# ── 1. 总资产校验 ────────────────────────────────────────────
|
||||
try:
|
||||
pf = mo_data.read_portfolio()
|
||||
mv_calc = sum(h["shares"] * h["price"] for h in pf.get("holdings", []) if h.get("price"))
|
||||
stored_ta = pf.get("total_assets", 0)
|
||||
cash = pf.get("cash", 0)
|
||||
|
||||
expected_ta = round(mv_calc + cash, 2)
|
||||
if stored_ta > 0 and abs(stored_ta - expected_ta) / stored_ta > 0.01:
|
||||
issues.append(f"总资产不匹配: 存储{stored_ta} ≠ 计算{expected_ta} (市值{mv_calc}+现金{cash})")
|
||||
except Exception as e:
|
||||
issues.append(f"portfolio.json读取失败: {e}")
|
||||
|
||||
# ── 2. 持仓 vs 决策交叉检查 ──────────────────────────────────
|
||||
try:
|
||||
dec = mo_data.read_decisions()
|
||||
dec_codes = {}
|
||||
for d in dec.get("decisions", []):
|
||||
dec_codes[d["code"]] = d
|
||||
|
||||
for h in pf.get("holdings", []):
|
||||
code = h["code"]
|
||||
if code not in dec_codes:
|
||||
issues.append(f"持仓{code}({h.get('name','?')}) 在decisions.json中无对应决策")
|
||||
|
||||
for code, d in dec_codes.items():
|
||||
if d.get("status") == "active" and d.get("type") == "持仓策略":
|
||||
if not any(h["code"] == code for h in pf.get("holdings", [])):
|
||||
issues.append(f"决策{code}({d.get('name','?')})标记持仓但portfolio.json中无此股")
|
||||
except Exception as e:
|
||||
issues.append(f"决策检查失败: {e}")
|
||||
|
||||
# ── 3. 币种一致性 ────────────────────────────────────────────
|
||||
try:
|
||||
for d in dec.get("decisions", []):
|
||||
code = str(d.get("code", ""))
|
||||
# 港股必须标记currency
|
||||
if len(code) == 5 and code[0] in ("0", "1"):
|
||||
cur = d.get("currency", "")
|
||||
if cur not in ("HKD", "CNY"):
|
||||
issues.append(f"港股{code}({d.get('name','?')}) 缺currency标记，不可靠")
|
||||
# 如果标记了HKD，stop_loss也应该是合理的HKD价（>10）
|
||||
sl = d.get("stop_loss", 0)
|
||||
if cur == "HKD" and sl > 0 and sl < 1:
|
||||
issues.append(f"港股{code} currency=HKD但stop_loss={sl} 异常低")
|
||||
except Exception as e:
|
||||
issues.append(f"币种检查失败: {e}")
|
||||
|
||||
# ── 4. 数据时效 ──────────────────────────────────────────────
|
||||
today = datetime.now().strftime("%Y-%m-%d")
|
||||
try:
|
||||
if pf.get("updated_at", "").startswith(today):
|
||||
pass # OK
|
||||
else:
|
||||
issues.append(f"portfolio.json updated_at={pf.get('updated_at','?')} 不是今日")
|
||||
except:
|
||||
pass
|
||||
|
||||
# ── 输出 ──────────────────────────────────────────────────────
|
||||
if issues:
|
||||
print("DATA_VALIDATE_FAIL")
|
||||
for i in issues:
|
||||
print(f" ⚠️ {i}")
|
||||
sys.exit(1)
|
||||
else:
|
||||
print("DATA_VALIDATE_OK")
|
||||
sys.exit(0)
|
||||
#!/usr/bin/env python3
|
||||
"""data_validate.py — 数据自检，在所有报告产出前执行
|
||||
|
||||
检查清单：
|
||||
1. 总资产 = 市值 + 现金 （误差 < 1%）
|
||||
2. 持仓 vs 决策交叉检查
|
||||
3. 币种一致性（港股必须currency=HKD）
|
||||
4. 数据时效：portfolio.json/decisions.json 今日已更新
|
||||
|
||||
返回值：通过→退出码0，输出"OK"。失败→退出码1，输出问题描述。
|
||||
"""
|
||||
|
||||
import json, sys
|
||||
from datetime import datetime, timezone
|
||||
from mo_data import read_portfolio, read_decisions, read_watchlist
|
||||
|
||||
DATA_DIR = "/home/hmo/web-dashboard/data"
|
||||
STALE_REPORT = f"{DATA_DIR}/strategy_staleness_report.json"
|
||||
|
||||
issues = []
|
||||
|
||||
# ── 1. 总资产校验 ────────────────────────────────────────────
|
||||
try:
|
||||
pf = mo_data.read_portfolio()
|
||||
mv_calc = sum(h["shares"] * h["price"] for h in pf.get("holdings", []) if h.get("price"))
|
||||
stored_ta = pf.get("total_assets", 0)
|
||||
cash = pf.get("cash", 0)
|
||||
|
||||
expected_ta = round(mv_calc + cash, 2)
|
||||
if stored_ta > 0 and abs(stored_ta - expected_ta) / stored_ta > 0.01:
|
||||
issues.append(f"总资产不匹配: 存储{stored_ta} ≠ 计算{expected_ta} (市值{mv_calc}+现金{cash})")
|
||||
except Exception as e:
|
||||
issues.append(f"portfolio.json读取失败: {e}")
|
||||
|
||||
# ── 2. 持仓 vs 决策交叉检查 ──────────────────────────────────
|
||||
try:
|
||||
dec = mo_data.read_decisions()
|
||||
dec_codes = {}
|
||||
for d in dec.get("decisions", []):
|
||||
dec_codes[d["code"]] = d
|
||||
|
||||
for h in pf.get("holdings", []):
|
||||
code = h["code"]
|
||||
if code not in dec_codes:
|
||||
issues.append(f"持仓{code}({h.get('name','?')}) 在decisions.json中无对应决策")
|
||||
|
||||
for code, d in dec_codes.items():
|
||||
if d.get("status") == "active" and d.get("type") == "持仓策略":
|
||||
if not any(h["code"] == code for h in pf.get("holdings", [])):
|
||||
issues.append(f"决策{code}({d.get('name','?')})标记持仓但portfolio.json中无此股")
|
||||
except Exception as e:
|
||||
issues.append(f"决策检查失败: {e}")
|
||||
|
||||
# ── 3. 币种一致性 ────────────────────────────────────────────
|
||||
try:
|
||||
for d in dec.get("decisions", []):
|
||||
code = str(d.get("code", ""))
|
||||
# 港股必须标记currency
|
||||
if len(code) == 5 and code[0] in ("0", "1"):
|
||||
cur = d.get("currency", "")
|
||||
if cur not in ("HKD", "CNY"):
|
||||
issues.append(f"港股{code}({d.get('name','?')}) 缺currency标记，不可靠")
|
||||
# 如果标记了HKD，stop_loss也应该是合理的HKD价（>10）
|
||||
sl = d.get("stop_loss", 0)
|
||||
if cur == "HKD" and sl > 0 and sl < 1:
|
||||
issues.append(f"港股{code} currency=HKD但stop_loss={sl} 异常低")
|
||||
except Exception as e:
|
||||
issues.append(f"币种检查失败: {e}")
|
||||
|
||||
# ── 4. 数据时效 ──────────────────────────────────────────────
|
||||
today = datetime.now().strftime("%Y-%m-%d")
|
||||
try:
|
||||
if pf.get("updated_at", "").startswith(today):
|
||||
pass # OK
|
||||
else:
|
||||
issues.append(f"portfolio.json updated_at={pf.get('updated_at','?')} 不是今日")
|
||||
except:
|
||||
pass
|
||||
|
||||
# ── 输出 ──────────────────────────────────────────────────────
|
||||
if issues:
|
||||
print("DATA_VALIDATE_FAIL")
|
||||
for i in issues:
|
||||
print(f" ⚠️ {i}")
|
||||
sys.exit(1)
|
||||
else:
|
||||
print("DATA_VALIDATE_OK")
|
||||
sys.exit(0)
|
||||
|
||||
Reference in New Issue
Block a user