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

@@ -334,9 +334,29 @@ def save_events(events):
 
 
 def record_event(code, name, event_type, price, trigger_value, event_label=""):
-    """记录一次价格触发事件到 price_events.json"""
+    """记录一次价格触发事件 — 双写：DB price_events 表（权威）+ price_events.json（遗留读取方兼容）"""
-    events = load_events()
     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:]

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()

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)

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)

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')

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')

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()