hmo/MoFin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: 同步4个库文件副本的提交内容与根canonical一致——消除deploy_guard回滚↔sync重链拉锯（guard实测抓到漂移并回滚，机制自证有效，但提交内容必须同源）

Browse Source
This commit is contained in:
hmo
2026-07-21 02:06:55 +08:00
parent 3e84cbc17e
commit beee453bf3
3 changed files with 177 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
deploy/profile-scripts/mofin_db.py
+88 -7
View File
@@ -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} 已写入"