diff --git a/mofin_db.py b/mofin_db.py index d84a3997..2a50f586 100644 --- a/mofin_db.py +++ b/mofin_db.py @@ -1367,20 +1367,56 @@ def flush_rec_digest(max_items=5): # 队列是打标瞬间的快照;flush 前回库读实时 信号/RR/tag, # 信号降级为弱信号或RR跌破2.0的条目直接丢弃——XMPP说的必须和盯盘一致。 import sqlite3 as _sq0 + from datetime import datetime as _ddt + _now = _ddt.now() + _h, _m, _w = _now.hour, _now.minute, _now.weekday() + _market_open = _w < 5 and ((_h == 9 and _m >= 30) or (10 <= _h < 15)) _vconn = _sq0.connect("/home/hmo/MoFin/data/mofin.db") _WEAK = ("信号不充分", "关注", "弱势持有", "观望", "持有", "") _live = [] for it in items: r = _vconn.execute( - "SELECT timing_signal, rr_ratio, tag FROM holding_strategies WHERE code=? AND status='active'", + "SELECT timing_signal, rr_ratio, tag, reassessed_at FROM holding_strategies WHERE code=? AND status='active'", (it['code'],)).fetchone() if not r: print(f" [REC] {it['code']} 已不在库,丢弃", flush=True) continue - cur_sig, cur_rr, cur_tag = r[0] or "", r[1] or 0, r[2] or "" + cur_sig, cur_rr, cur_tag, cur_ra = r[0] or "", r[1] or 0, r[2] or "", r[3] or "" if cur_tag != 'current_recommend': print(f" [REC] {it['code']} tag已撤销({cur_tag}),丢弃", flush=True) continue + # ── 数据时效校验(2026-07-27 老爸:盘前分析盘中推送=过期数据误导)── + # 分析时间在开盘前且现在已开盘 → 触发盘中重评(用实时数据分析,不推旧分析) + if _market_open and cur_ra: + try: + _ra_dt = _ddt.fromisoformat(str(cur_ra)[:19]) + # 分析在 08:00-09:29 之间做的 = 盘前分析 → 盘中触发重评 + if (_ra_dt.hour >= 8 and (_ra_dt.hour < 9 or (_ra_dt.hour == 9 and _ra_dt.minute < 30))) \ + and (_ddt.now() - _ra_dt).total_seconds() > 120: + try: + import subprocess as _sp + _re = _sp.run( + ["python3", "/home/hmo/MoFin/deploy/profile-scripts/per_stock_reassess.py", it['code']], + capture_output=True, text=True, timeout=90) + if _re.returncode == 0: + print(f" [REC] {it['code']} 盘中重评完成,用实时数据更新策略", flush=True) + # 重新读库获取更新后数据 + r = _vconn.execute( + "SELECT timing_signal, rr_ratio, tag FROM holding_strategies WHERE code=? AND status='active'", + (it['code'],)).fetchone() + if r: + cur_sig, cur_rr, cur_tag = r[0] or "", r[1] or 0, r[2] or "" + if cur_tag != 'current_recommend': + print(f" [REC] {it['code']} 重评后tag撤销,丢弃", flush=True) + continue + else: + print(f" [REC] {it['code']} 盘中重评失败(rc={_re.returncode}),标记警告", flush=True) + it['_stale_warn'] = "⚠️ 盘前分析(已开盘未能及时重评,请结合实时盘面判断)" + except subprocess.TimeoutExpired: + print(f" [REC] {it['code']} 盘中重评超时,标记警告", flush=True) + it['_stale_warn'] = "⚠️ 盘前分析(已开盘未能及时重评,请结合实时盘面判断)" + except Exception: + pass if it.get('signal') in ("买入", "可买入", "可加仓"): if cur_sig in _WEAK: print(f" [REC] {it['code']} 信号降级为'{cur_sig}',丢弃", flush=True) @@ -1480,7 +1516,9 @@ def flush_rec_digest(max_items=5): f" 区{it.get('entry_low') or '—'}~{it.get('entry_high') or '—'}" f" 损{it.get('stop_loss') or '—'} 盈{it.get('take_profit') or '—'}" f" {_rr_txt} 仓位{it.get('position') or '—'}") - # 头部 2 只附策略依据(12维全文节选) + if it.get('_stale_warn'): + lines.append(f" ⚠️ {it['_stale_warn']}") + # 头部 2 只附策略依据(12维全文节选) if i < 2: if it.get('strategy_excerpt'): lines.append(f" 依据: {it['strategy_excerpt'][:300]}")