diff --git a/deploy/profile-scripts/mo_data.py b/deploy/profile-scripts/mo_data.py index db6c7e01..cb168a13 100644 --- a/deploy/profile-scripts/mo_data.py +++ b/deploy/profile-scripts/mo_data.py @@ -82,7 +82,7 @@ def read_decisions(): "created_at, updated_at, " "avg_price, decision_timestamp, note, quality_check, " "quality_checked_at, quality_issues_json, position_advice, " - "signal_factors_json, time_horizon, decision_type " + "signal_factors_json, time_horizon, decision_type, tag " "FROM holding_strategies WHERE status IN ('active','updated') " "ORDER BY code" ).fetchall() diff --git a/deploy/profile-scripts/mofin_db.py b/deploy/profile-scripts/mofin_db.py index 9cdec1e9..98ff7177 100644 --- a/deploy/profile-scripts/mofin_db.py +++ b/deploy/profile-scripts/mofin_db.py @@ -262,6 +262,29 @@ def init_all_tables(conn: sqlite3.Connection): CREATE UNIQUE INDEX IF NOT EXISTS idx_strategy_code ON holding_strategies(code); CREATE INDEX IF NOT EXISTS idx_strategy_status ON holding_strategies(status); + -- 策略历史快照（每次覆写前自动记录） + CREATE TABLE IF NOT EXISTS strategy_history ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + code TEXT NOT NULL, + name TEXT, + decision_type TEXT, + strategy_type TEXT, + full_analysis TEXT, + action TEXT, + timing_signal TEXT, + entry_low REAL, + entry_high REAL, + stop_loss REAL, + take_profit REAL, + position_advice TEXT, + rr_ratio REAL, + version INTEGER, + source_trigger TEXT, + reassessed_at TEXT, + snapshotted_at TEXT NOT NULL + ); + CREATE INDEX IF NOT EXISTS idx_strategy_history_code ON strategy_history(code, snapshotted_at); + -- 自选股 CREATE TABLE IF NOT EXISTS watchlist_stocks ( code TEXT PRIMARY KEY REFERENCES stocks(code), @@ -551,6 +574,14 @@ def init_all_tables(conn: sqlite3.Connection): conn.execute(f"ALTER TABLE {table} ADD COLUMN {col_def}") except sqlite3.OperationalError: pass # column already exists + + # ── tag 迁移（2026-07-20）：推荐标签 current_recommend / active_manual ── + # 此前 strategy_lifecycle 在 dict 里设置 tag 但 write_holding_strategy 无此列， + # 导致标签在写入时被静默丢弃。补列 + 写入保留。 + try: + conn.execute("ALTER TABLE holding_strategies ADD COLUMN tag TEXT DEFAULT ''") + except sqlite3.OperationalError: + pass conn.commit() @@ -1077,9 +1108,52 @@ def get_prices_batch_from_db(codes: list[str]) -> dict: # 核心写函数 — 替代 json.dump()，强制币种约束 # ═══════════════════════════════════════════════════════════════════ -def write_holding_strategy(conn, code: str, name: str, data: dict) -> tuple[bool, str]: +def snapshot_strategy_history(conn, code: str, source_trigger: str = "write_holding_strategy"): + """在修改前快照当前策略到 strategy_history 表。永不抛异常。""" + try: + row = conn.execute( + "SELECT code, name, decision_type, strategy_type, full_analysis, " + "action, timing_signal, entry_low, entry_high, stop_loss, take_profit, " + "position_advice, rr_ratio, version, reassessed_at " + "FROM holding_strategies WHERE code=? AND status='active'", + (code,) + ).fetchone() + if not row: + return + now = datetime.now().isoformat() + conn.execute(""" + INSERT INTO strategy_history + (code, name, decision_type, strategy_type, full_analysis, action, + timing_signal, entry_low, entry_high, stop_loss, take_profit, + position_advice, rr_ratio, version, source_trigger, reassessed_at, snapshotted_at) + VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) + """, ( + row[0], row[1], row[2], row[3], + row[4], row[5], row[6], + row[7], row[8], row[9], row[10], + row[11], row[12], row[13], + source_trigger, row[14], now + )) + conn.commit() + # 每只股票只保留最近20条历史 + conn.execute(""" + DELETE FROM strategy_history WHERE code=? AND id NOT IN ( + SELECT id FROM strategy_history WHERE code=? ORDER BY snapshotted_at DESC LIMIT 20 + ) + """, (code, code)) + conn.commit() + except Exception as e: + print(f" [SNAPSHOT] {code} 快照失败: {e}", flush=True) + + +def write_holding_strategy(conn, code: str, name: str, data: dict, + source_trigger: str = "write_holding_strategy") -> tuple[bool, str]: """写入持仓策略（替代 decisions.json 单条写入）。data 必须包含 currency。""" try: + # ── 覆写前快照旧行 ── + snapshot_strategy_history(conn, code, source_trigger) + + currency = data.get('currency', 'CNY') # Serialize JSON fields import json as _json @@ -1091,12 +1165,18 @@ def write_holding_strategy(conn, code: str, name: str, data: dict) -> tuple[bool # 在DELETE前保留现有的full_analysis和reassessed_at（防止被regenerate_all等清空） _existing_fa = data.get('full_analysis', '') _existing_ra = data.get('reassessed_at', '') - if not _existing_fa: + # tag 语义：'tag' 键缺席=保留旧标签；显式传入（含''）= 按传入值（允许清除标签） + _tag_absent = 'tag' not in data + _existing_tag = data.get('tag', '') or '' + if not _existing_fa or _tag_absent: try: - _old = conn.execute("SELECT full_analysis, reassessed_at FROM holding_strategies WHERE code=? ORDER BY id DESC LIMIT 1", (code,)).fetchone() + _old = conn.execute("SELECT full_analysis, reassessed_at, tag FROM holding_strategies WHERE code=? ORDER BY id DESC LIMIT 1", (code,)).fetchone() if _old: - if _old[0]: _existing_fa = _old[0] - if _old[1]: _existing_ra = _old[1] + if not _existing_fa: + if _old[0]: _existing_fa = _old[0] + if _old[1]: _existing_ra = _old[1] + if _tag_absent and _old[2]: + _existing_tag = _old[2] except: pass @@ -1112,10 +1192,10 @@ def write_holding_strategy(conn, code: str, name: str, data: dict) -> tuple[bool avg_price, decision_timestamp, note, quality_check, quality_checked_at, quality_issues_json, position_advice, signal_factors_json, time_horizon, decision_type, - full_analysis, reassessed_at) + full_analysis, reassessed_at, tag) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?, datetime('now','localtime'), - ?,?,?,?,?,?,?,?,?,?,?,?) + ?,?,?,?,?,?,?,?,?,?,?,?,?) """, ( code, name, data.get('version', 1), data.get('price'), data.get('cost'), @@ -1141,6 +1221,7 @@ def write_holding_strategy(conn, code: str, name: str, data: dict) -> tuple[bool # 保留full_analysis和reassessed_at _existing_fa, _existing_ra, + _existing_tag, )) conn.commit() return True, f"策略 {code} 已写入" diff --git a/scripts/mofin_db.py b/scripts/mofin_db.py index 9cdec1e9..98ff7177 100644 --- a/scripts/mofin_db.py +++ b/scripts/mofin_db.py @@ -262,6 +262,29 @@ def init_all_tables(conn: sqlite3.Connection): CREATE UNIQUE INDEX IF NOT EXISTS idx_strategy_code ON holding_strategies(code); CREATE INDEX IF NOT EXISTS idx_strategy_status ON holding_strategies(status); + -- 策略历史快照（每次覆写前自动记录） + CREATE TABLE IF NOT EXISTS strategy_history ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + code TEXT NOT NULL, + name TEXT, + decision_type TEXT, + strategy_type TEXT, + full_analysis TEXT, + action TEXT, + timing_signal TEXT, + entry_low REAL, + entry_high REAL, + stop_loss REAL, + take_profit REAL, + position_advice TEXT, + rr_ratio REAL, + version INTEGER, + source_trigger TEXT, + reassessed_at TEXT, + snapshotted_at TEXT NOT NULL + ); + CREATE INDEX IF NOT EXISTS idx_strategy_history_code ON strategy_history(code, snapshotted_at); + -- 自选股 CREATE TABLE IF NOT EXISTS watchlist_stocks ( code TEXT PRIMARY KEY REFERENCES stocks(code), @@ -551,6 +574,14 @@ def init_all_tables(conn: sqlite3.Connection): conn.execute(f"ALTER TABLE {table} ADD COLUMN {col_def}") except sqlite3.OperationalError: pass # column already exists + + # ── tag 迁移（2026-07-20）：推荐标签 current_recommend / active_manual ── + # 此前 strategy_lifecycle 在 dict 里设置 tag 但 write_holding_strategy 无此列， + # 导致标签在写入时被静默丢弃。补列 + 写入保留。 + try: + conn.execute("ALTER TABLE holding_strategies ADD COLUMN tag TEXT DEFAULT ''") + except sqlite3.OperationalError: + pass conn.commit() @@ -1077,9 +1108,52 @@ def get_prices_batch_from_db(codes: list[str]) -> dict: # 核心写函数 — 替代 json.dump()，强制币种约束 # ═══════════════════════════════════════════════════════════════════ -def write_holding_strategy(conn, code: str, name: str, data: dict) -> tuple[bool, str]: +def snapshot_strategy_history(conn, code: str, source_trigger: str = "write_holding_strategy"): + """在修改前快照当前策略到 strategy_history 表。永不抛异常。""" + try: + row = conn.execute( + "SELECT code, name, decision_type, strategy_type, full_analysis, " + "action, timing_signal, entry_low, entry_high, stop_loss, take_profit, " + "position_advice, rr_ratio, version, reassessed_at " + "FROM holding_strategies WHERE code=? AND status='active'", + (code,) + ).fetchone() + if not row: + return + now = datetime.now().isoformat() + conn.execute(""" + INSERT INTO strategy_history + (code, name, decision_type, strategy_type, full_analysis, action, + timing_signal, entry_low, entry_high, stop_loss, take_profit, + position_advice, rr_ratio, version, source_trigger, reassessed_at, snapshotted_at) + VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) + """, ( + row[0], row[1], row[2], row[3], + row[4], row[5], row[6], + row[7], row[8], row[9], row[10], + row[11], row[12], row[13], + source_trigger, row[14], now + )) + conn.commit() + # 每只股票只保留最近20条历史 + conn.execute(""" + DELETE FROM strategy_history WHERE code=? AND id NOT IN ( + SELECT id FROM strategy_history WHERE code=? ORDER BY snapshotted_at DESC LIMIT 20 + ) + """, (code, code)) + conn.commit() + except Exception as e: + print(f" [SNAPSHOT] {code} 快照失败: {e}", flush=True) + + +def write_holding_strategy(conn, code: str, name: str, data: dict, + source_trigger: str = "write_holding_strategy") -> tuple[bool, str]: """写入持仓策略（替代 decisions.json 单条写入）。data 必须包含 currency。""" try: + # ── 覆写前快照旧行 ── + snapshot_strategy_history(conn, code, source_trigger) + + currency = data.get('currency', 'CNY') # Serialize JSON fields import json as _json @@ -1091,12 +1165,18 @@ def write_holding_strategy(conn, code: str, name: str, data: dict) -> tuple[bool # 在DELETE前保留现有的full_analysis和reassessed_at（防止被regenerate_all等清空） _existing_fa = data.get('full_analysis', '') _existing_ra = data.get('reassessed_at', '') - if not _existing_fa: + # tag 语义：'tag' 键缺席=保留旧标签；显式传入（含''）= 按传入值（允许清除标签） + _tag_absent = 'tag' not in data + _existing_tag = data.get('tag', '') or '' + if not _existing_fa or _tag_absent: try: - _old = conn.execute("SELECT full_analysis, reassessed_at FROM holding_strategies WHERE code=? ORDER BY id DESC LIMIT 1", (code,)).fetchone() + _old = conn.execute("SELECT full_analysis, reassessed_at, tag FROM holding_strategies WHERE code=? ORDER BY id DESC LIMIT 1", (code,)).fetchone() if _old: - if _old[0]: _existing_fa = _old[0] - if _old[1]: _existing_ra = _old[1] + if not _existing_fa: + if _old[0]: _existing_fa = _old[0] + if _old[1]: _existing_ra = _old[1] + if _tag_absent and _old[2]: + _existing_tag = _old[2] except: pass @@ -1112,10 +1192,10 @@ def write_holding_strategy(conn, code: str, name: str, data: dict) -> tuple[bool avg_price, decision_timestamp, note, quality_check, quality_checked_at, quality_issues_json, position_advice, signal_factors_json, time_horizon, decision_type, - full_analysis, reassessed_at) + full_analysis, reassessed_at, tag) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?, datetime('now','localtime'), - ?,?,?,?,?,?,?,?,?,?,?,?) + ?,?,?,?,?,?,?,?,?,?,?,?,?) """, ( code, name, data.get('version', 1), data.get('price'), data.get('cost'), @@ -1141,6 +1221,7 @@ def write_holding_strategy(conn, code: str, name: str, data: dict) -> tuple[bool # 保留full_analysis和reassessed_at _existing_fa, _existing_ra, + _existing_tag, )) conn.commit() return True, f"策略 {code} 已写入"