From efdfaf956aac94c55cbab95f126558c79e70acc9 Mon Sep 17 00:00:00 2001 From: hmo Date: Mon, 20 Jul 2026 17:46:57 +0800 Subject: [PATCH] fix(price_events): unify event storage to DB (dual-write + backfill) User caught the inconsistency: system claims DB-first but price events only went to price_events.json, leaving DB table stale since Jul 6. Root cause chain found: - record_event() only wrote JSON, never called mofin_db.write_price_event - price_events.code has FK -> stocks(code); events for unregistered stocks (new candidates, HK) silently failed INSERT and were lost to DB - mofin_db.write_price_event swallows errors (returns False silently) Fixes: - record_event now dual-writes: DB (authoritative) + JSON (compat for legacy readers mo_config/strategy_feedback/system_health_check) - auto-registers unknown codes into stocks table before event insert - one-time backfill: 4064 JSON events -> DB (total 6353 rows, last=today) - verified: record_event TEST99 lands in both DB and JSON --- deploy/profile-scripts/price_monitor.py | 26 +++++++++++-- scripts/backfill_price_events.py | 50 +++++++++++++++++++++++++ scripts/check_fk.py | 7 ++++ scripts/check_stocks_table.py | 9 +++++ scripts/test_db_write.py | 15 ++++++++ scripts/test_dual_write.py | 23 ++++++++++++ scripts/test_raw_insert.py | 20 ++++++++++ 7 files changed, 147 insertions(+), 3 deletions(-) create mode 100644 scripts/backfill_price_events.py create mode 100644 scripts/check_fk.py create mode 100644 scripts/check_stocks_table.py create mode 100644 scripts/test_db_write.py create mode 100644 scripts/test_dual_write.py create mode 100644 scripts/test_raw_insert.py diff --git a/deploy/profile-scripts/price_monitor.py b/deploy/profile-scripts/price_monitor.py index 5d23c525..7d7fa0f6 100644 --- a/deploy/profile-scripts/price_monitor.py +++ b/deploy/profile-scripts/price_monitor.py @@ -334,9 +334,29 @@ def save_events(events): def record_event(code, name, event_type, price, trigger_value, event_label=""): - """记录一次价格触发事件到 price_events.json""" - events = load_events() + """记录一次价格触发事件 — 双写：DB price_events 表（权威）+ price_events.json（遗留读取方兼容）""" now = datetime.now().isoformat() + date_str = datetime.now().strftime("%Y-%m-%d") + + # 1. 写 DB 表（权威存储）。price_events.code 有 FK -> stocks(code)， + # 未注册的股票（新候选/港股）会先注册再写事件，否则 FK 失败事件丢失。 + if HAS_DB: + try: + from mofin_db import get_conn, write_price_event + _c = get_conn() + _exch, _typ = ("HK", "H") if len(str(code)) == 5 else (("SH", "A") if str(code).startswith(("6", "9")) else ("SZ", "A")) + _c.execute("INSERT OR IGNORE INTO stocks (code, name, exchange, type, updated_at) VALUES (?,?,?,?,?)", + (str(code), name or str(code), _exch, _typ, now)) + _c.commit() + write_price_event(_c, code=code, name=name, event_type=event_type, + price=round(price, 2), trigger_value=str(trigger_value), + event_label=event_label) + _c.close() + except Exception as e: + print(f"[price_events DB写入失败] {e}", file=sys.stderr) + + # 2. 写 JSON（遗留读取方：mo_config/strategy_feedback/system_health_check 还在读） + events = load_events() events["events"].append({ "code": code, "name": name, @@ -345,7 +365,7 @@ def record_event(code, name, event_type, price, trigger_value, event_label=""): "trigger_value": trigger_value, "event_label": event_label, "timestamp": now, - "date": datetime.now().strftime("%Y-%m-%d"), + "date": date_str, }) # 保留最近10000条 events["events"] = events["events"][-10000:] diff --git a/scripts/backfill_price_events.py b/scripts/backfill_price_events.py new file mode 100644 index 00000000..be26f692 --- /dev/null +++ b/scripts/backfill_price_events.py @@ -0,0 +1,50 @@ +import json, sqlite3 +from datetime import datetime + +d = json.load(open('/home/hmo/web-dashboard/data/price_events.json')) +events = d.get('events', d if isinstance(d, list) else []) +print(f'JSON events: {len(events)}') + +conn = sqlite3.connect('/home/hmo/MoFin/data/mofin.db', timeout=30) +conn.execute('PRAGMA busy_timeout=30000') + +# 已有的 DB 记录（避免重复）：用 code+created_at 做粗粒度去重 +existing = set() +for r in conn.execute("SELECT code, created_at FROM price_events"): + existing.add((r[0], r[1])) + +inserted = 0 +skipped = 0 +for e in events: + code = str(e.get('code', '')) + name = e.get('name', code) + ts = e.get('timestamp') or e.get('created_at') or '' + if not code or not ts: + skipped += 1 + continue + # 统一时间格式 + try: + dt = datetime.fromisoformat(str(ts).replace('Z', '')) + created = dt.strftime('%Y-%m-%d %H:%M:%S') + date = dt.strftime('%Y-%m-%d') + except Exception: + skipped += 1 + continue + if (code, created) in existing: + skipped += 1 + continue + exch, typ = ('HK', 'H') if len(code) == 5 else (('SH', 'A') if code.startswith(('6', '9')) else ('SZ', 'A')) + conn.execute("INSERT OR IGNORE INTO stocks (code, name, exchange, type, updated_at) VALUES (?,?,?,?,?)", + (code, name, exch, typ, created)) + conn.execute( + "INSERT INTO price_events (code, name, event_type, price, trigger_value, event_label, created_at, date) " + "VALUES (?,?,?,?,?,?,?,?)", + (code, name, e.get('event_type', ''), e.get('price', 0), + str(e.get('trigger_value', '')), e.get('event_label', ''), created, date)) + inserted += 1 + +conn.commit() +total = conn.execute("SELECT COUNT(*) FROM price_events").fetchone()[0] +last = conn.execute("SELECT MAX(created_at) FROM price_events").fetchone()[0] +print(f'inserted={inserted} skipped={skipped} total={total} last={last}') +conn.close() \ No newline at end of file diff --git a/scripts/check_fk.py b/scripts/check_fk.py new file mode 100644 index 00000000..5ef94e24 --- /dev/null +++ b/scripts/check_fk.py @@ -0,0 +1,7 @@ +import sqlite3 +c = sqlite3.connect('/home/hmo/MoFin/data/mofin.db') +sql = c.execute("SELECT sql FROM sqlite_master WHERE name='price_events'").fetchone()[0] +print(sql) +print() +for r in c.execute("PRAGMA foreign_key_list(price_events)"): + print(r) \ No newline at end of file diff --git a/scripts/check_stocks_table.py b/scripts/check_stocks_table.py new file mode 100644 index 00000000..fc78817d --- /dev/null +++ b/scripts/check_stocks_table.py @@ -0,0 +1,9 @@ +import sqlite3 +c = sqlite3.connect('/home/hmo/MoFin/data/mofin.db') +sql = c.execute("SELECT sql FROM sqlite_master WHERE name='stocks'").fetchone() +print(sql[0] if sql else 'no stocks table') +print() +cnt = c.execute("SELECT COUNT(*) FROM stocks").fetchone()[0] +print('rows:', cnt) +for r in c.execute("SELECT * FROM stocks LIMIT 3"): + print(r) \ No newline at end of file diff --git a/scripts/test_db_write.py b/scripts/test_db_write.py new file mode 100644 index 00000000..c3af02db --- /dev/null +++ b/scripts/test_db_write.py @@ -0,0 +1,15 @@ +import sys +sys.path.insert(0, '/home/hmo/.hermes/profiles/position-analyst/scripts') +sys.path.insert(0, '/home/hmo/MoFin') +from mofin_db import get_conn, write_price_event + +c = get_conn() +ok = write_price_event(c, code='TEST99', name='测试股', event_type='entry_zone', + price=12.34, trigger_value='12.0~12.5', event_label='加仓区间') +print('write_price_event returned:', ok) +r = c.execute("SELECT code,created_at FROM price_events WHERE code='TEST99'").fetchone() +print('row:', r) +c.execute("DELETE FROM price_events WHERE code='TEST99'") +c.commit() +c.close() +print('done') \ No newline at end of file diff --git a/scripts/test_dual_write.py b/scripts/test_dual_write.py new file mode 100644 index 00000000..10d79c99 --- /dev/null +++ b/scripts/test_dual_write.py @@ -0,0 +1,23 @@ +import sys +sys.path.insert(0, '/home/hmo/.hermes/profiles/position-analyst/scripts') +import price_monitor as pm +pm.record_event('TEST99', '测试股', 'entry_zone', 12.34, '12.0~12.5', '加仓区间') +print('record_event OK') + +import sqlite3 +c = sqlite3.connect('/home/hmo/MoFin/data/mofin.db') +r = c.execute("SELECT code,name,event_type,price,created_at FROM price_events WHERE code='TEST99' ORDER BY id DESC LIMIT 1").fetchone() +print('DB row:', r) +c.execute("DELETE FROM price_events WHERE code='TEST99'") +c.commit() +print('cleaned') + +# also check JSON got it +import json +d = json.load(open('/home/hmo/web-dashboard/data/price_events.json')) +last = d['events'][-1] +print('JSON last:', last['code'], last['price']) +if last['code'] == 'TEST99': + d['events'] = d['events'][:-1] + json.dump(d, open('/home/hmo/web-dashboard/data/price_events.json', 'w'), ensure_ascii=False, indent=2) + print('JSON cleaned') \ No newline at end of file diff --git a/scripts/test_raw_insert.py b/scripts/test_raw_insert.py new file mode 100644 index 00000000..e93307d1 --- /dev/null +++ b/scripts/test_raw_insert.py @@ -0,0 +1,20 @@ +import sys, sqlite3, traceback +sys.path.insert(0, '/home/hmo/MoFin') +from mofin_db import get_conn, DB_PATH +print('DB_PATH:', DB_PATH) + +c = get_conn() +try: + c.execute( + "INSERT INTO price_events (code, name, event_type, price, trigger_value, event_label, date) " + "VALUES (?, ?, ?, ?, ?, ?, ?)", + ('TEST99', '测试股', 'entry_zone', 12.34, '12.0~12.5', '加仓区间', '2026-07-20')) + c.commit() + print('INSERT OK') +except Exception as e: + traceback.print_exc() +r = c.execute("SELECT code FROM price_events WHERE code='TEST99'").fetchone() +print('row:', r) +c.execute("DELETE FROM price_events WHERE code='TEST99'") +c.commit() +c.close() \ No newline at end of file