修复复盘列表与统计并行同步撞锁导致模拟盘加载失败。

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
dekun
2026-10-07 13:46:38 +08:00
parent cfcc0596ff
commit 2158a30179
5 changed files with 83 additions and 27 deletions
+4 -1
View File
@@ -929,7 +929,10 @@
var kpi = $("or-kpi"); var kpi = $("or-kpi");
var groups = $("or-stats-groups"); var groups = $("or-stats-groups");
if (!kpi || !groups) return; if (!kpi || !groups) return;
fetch("/api/options/review/stats?" + baseQs().toString(), { credentials: "same-origin" }) // 与 trades?sync=1 并行时不再二次同步,避免 SQLite database is locked
var qs = baseQs();
qs.set("sync", "0");
fetch("/api/options/review/stats?" + qs.toString(), { credentials: "same-origin" })
.then(function (r) { .then(function (r) {
return r.json(); return r.json();
}) })
+32 -14
View File
@@ -66,21 +66,28 @@ def _safe_float(v: Any) -> Optional[float]:
def get_sync_state(conn: sqlite3.Connection, key: str) -> Optional[str]: def get_sync_state(conn: sqlite3.Connection, key: str) -> Optional[str]:
row = conn.execute( try:
"SELECT value FROM options_review_sync_state WHERE key=?", (key,) row = conn.execute(
).fetchone() "SELECT value FROM options_review_sync_state WHERE key=?", (key,)
).fetchone()
except sqlite3.OperationalError:
return None
return str(row["value"]) if row and row["value"] is not None else None return str(row["value"]) if row and row["value"] is not None else None
def set_sync_state(conn: sqlite3.Connection, key: str, value: str) -> None: def set_sync_state(conn: sqlite3.Connection, key: str, value: str) -> None:
conn.execute( try:
""" conn.execute(
INSERT INTO options_review_sync_state(key, value, updated_at) """
VALUES (?, ?, ?) INSERT INTO options_review_sync_state(key, value, updated_at)
ON CONFLICT(key) DO UPDATE SET value=excluded.value, updated_at=excluded.updated_at VALUES (?, ?, ?)
""", ON CONFLICT(key) DO UPDATE SET value=excluded.value, updated_at=excluded.updated_at
(key, value, _now_str()), """,
) (key, value, _now_str()),
)
except sqlite3.OperationalError:
# 列表/统计并发刷新时可能撞锁;标记失败不阻断读列表
pass
def _purge_review_trade_by_key(conn: sqlite3.Connection, history_key: str) -> bool: def _purge_review_trade_by_key(conn: sqlite3.Connection, history_key: str) -> bool:
@@ -627,8 +634,16 @@ def ensure_local_review_synced(
repair_swapped_closed_at_signal_note, repair_swapped_closed_at_signal_note,
) )
ts_migrate = migrate_options_utc_wall_clock_to_app(conn) ts_migrate: dict[str, Any] = {}
note_swap = repair_swapped_closed_at_signal_note(conn) note_swap: dict[str, Any] = {}
try:
ts_migrate = migrate_options_utc_wall_clock_to_app(conn)
except Exception as e:
ts_migrate = {"ok": False, "msg": str(e)}
try:
note_swap = repair_swapped_closed_at_signal_note(conn)
except Exception as e:
note_swap = {"ok": False, "msg": str(e)}
if backfill_exchange_pnl and ex is not None: if backfill_exchange_pnl and ex is not None:
try: try:
from lib.exchange.okx_options_lib import fetch_all_option_positions_history from lib.exchange.okx_options_lib import fetch_all_option_positions_history
@@ -640,7 +655,10 @@ def ensure_local_review_synced(
backfill_closed_options_realized_pnl_from_history(conn, hist) backfill_closed_options_realized_pnl_from_history(conn, hist)
except Exception: except Exception:
pass pass
out = sync_all_review_sources(conn, ex=ex, from_exchange=False) try:
out = sync_all_review_sources(conn, ex=ex, from_exchange=False)
except sqlite3.OperationalError as e:
out = {"ok": False, "msg": f"sync locked: {e}"}
if isinstance(out, dict): if isinstance(out, dict):
out["ts_migrate"] = ts_migrate out["ts_migrate"] = ts_migrate
out["note_swap"] = note_swap out["note_swap"] = note_swap
+32 -6
View File
@@ -291,9 +291,18 @@ def register_options_review_routes(app: Flask, cfg: dict[str, Any], repo_root: s
"no", "no",
) )
if do_sync: if do_sync:
ex, _err = _require_ex(cfg) try:
ensure_local_review_synced(conn, ex=ex if ex is not None else None) ex, _err = _require_ex(cfg)
conn.commit() ensure_local_review_synced(conn, ex=ex if ex is not None else None)
conn.commit()
except Exception as sync_err:
# 同步撞锁时仍返回已有列表,避免整页 500
print(f"[options_review_trades] sync skipped: {sync_err}", flush=True)
try:
conn.rollback()
except Exception:
pass
init_options_review_tables(conn)
filt = dict( filt = dict(
source_type=_review_source_for_mode(request.args.get("source_type")), source_type=_review_source_for_mode(request.args.get("source_type")),
underlying=(request.args.get("underlying") or "").strip() or None, underlying=(request.args.get("underlying") or "").strip() or None,
@@ -434,9 +443,26 @@ def register_options_review_routes(app: Flask, cfg: dict[str, Any], repo_root: s
def api_options_review_stats(): def api_options_review_stats():
conn = cfg["get_db"]() conn = cfg["get_db"]()
try: try:
ex, _err = _require_ex(cfg) # 默认不同步:与 trades?sync=1 并行时避免双写撞锁;需要时再 sync=1
ensure_local_review_synced(conn, ex=ex if ex is not None else None) do_sync = (request.args.get("sync") or "0").strip().lower() in (
conn.commit() "1",
"true",
"yes",
)
if do_sync:
try:
ex, _err = _require_ex(cfg)
ensure_local_review_synced(conn, ex=ex if ex is not None else None)
conn.commit()
except Exception as sync_err:
print(f"[options_review_stats] sync skipped: {sync_err}", flush=True)
try:
conn.rollback()
except Exception:
pass
init_options_review_tables(conn)
else:
init_options_review_tables(conn)
stats = compute_review_stats( stats = compute_review_stats(
conn, conn,
source_type=_review_source_for_mode(request.args.get("source_type")), source_type=_review_source_for_mode(request.args.get("source_type")),
+14 -5
View File
@@ -112,11 +112,16 @@ def migrate_options_utc_wall_clock_to_app(conn: sqlite3.Connection) -> dict[str,
init_options_tables(conn) init_options_tables(conn)
trades = _shift_table("options_trades", ("created_at", "closed_at")) trades = _shift_table("options_trades", ("created_at", "closed_at"))
except sqlite3.OperationalError:
return {"skipped": 0, "trades": 0, "review": 0, "hidden": 0, "locked": 1}
except Exception: except Exception:
trades = 0 trades = 0
review = _shift_table("options_review_trades", ("opened_at", "closed_at")) try:
hidden = _shift_table("options_review_hidden", ("closed_at",)) review = _shift_table("options_review_trades", ("opened_at", "closed_at"))
hidden = _shift_table("options_review_hidden", ("closed_at",))
except sqlite3.OperationalError:
return {"skipped": 0, "trades": trades, "review": 0, "hidden": 0, "locked": 1}
set_sync_state(conn, _MIGRATE_FLAG, "1") set_sync_state(conn, _MIGRATE_FLAG, "1")
return {"skipped": 0, "trades": trades, "review": review, "hidden": hidden} return {"skipped": 0, "trades": trades, "review": review, "hidden": hidden}
@@ -129,7 +134,7 @@ def repair_swapped_closed_at_signal_note(conn: sqlite3.Connection) -> dict[str,
幂等:同步状态 options_closed_at_note_swap_done=1 后仍扫描可修行(新坏行可再修). 幂等:同步状态 options_closed_at_note_swap_done=1 后仍扫描可修行(新坏行可再修).
""" """
from lib.options.options_review_db import init_options_review_tables from lib.options.options_review_db import init_options_review_tables
from lib.options.options_review_lib import set_sync_state from lib.options.options_review_lib import get_sync_state, set_sync_state
init_options_review_tables(conn) init_options_review_tables(conn)
trades_fixed = 0 trades_fixed = 0
@@ -174,9 +179,13 @@ def repair_swapped_closed_at_signal_note(conn: sqlite3.Connection) -> dict[str,
(note, hk, closed), (note, hk, closed),
) )
review_fixed += int(cur.rowcount or 0) review_fixed += int(cur.rowcount or 0)
except sqlite3.OperationalError:
# 并发刷新撞锁时跳过本轮修复,不影响列表读取
return {"trades": 0, "review": 0, "locked": 1}
except Exception: except Exception:
pass pass
# 复盘表里仍残留非时间戳 closed_at 的,若能从 local_opt 源修好则已处理;否则仅标记完成 # 已标记过则不再写同步状态,减少并发写冲突
set_sync_state(conn, _SWAP_REPAIR_FLAG, "1") if get_sync_state(conn, _SWAP_REPAIR_FLAG) != "1":
set_sync_state(conn, _SWAP_REPAIR_FLAG, "1")
return {"trades": trades_fixed, "review": review_fixed} return {"trades": trades_fixed, "review": review_fixed}
@@ -456,4 +456,4 @@
</section> </section>
</div> </div>
<script src="/static/options_review.js?v=24"></script> <script src="/static/options_review.js?v=25"></script>