修复期权开平仓时间按 UTC 入库导致复盘K线标记偏8小时：统一业务时区并一次性迁移。

Co-authored-by: Cursor <cursoragent@cursor.com>

lib/exchange/okx_options_lib.py

@@ -214,16 +214,10 @@ def option_history_row_key(
 
 
 def _ms_to_iso(ms: Any) -> str | None:
-    val = _safe_float(ms)
+    """OKX ms → 业务时区墙钟(默认 Asia/Shanghai);勿用系统本地时区."""
-    if val is None or val <= 0:
+    from lib.options.options_time_lib import ms_to_options_app_str
-        return None
-    try:
-        from datetime import datetime, timezone
 
-        dt = datetime.fromtimestamp(int(val) / 1000.0, tz=timezone.utc).astimezone()
+    return ms_to_options_app_str(ms)
-        return dt.strftime("%Y-%m-%d %H:%M:%S")
-    except (TypeError, ValueError, OSError):
-        return None
 
 
 def option_instrument_meta_cached(

lib/options/options_close_exec_lib.py

@@ -361,12 +361,15 @@ def close_option_by_bid1(
                         ELSE signal_note
                     END"""
                     params.append(signal_note)
+                from lib.options.options_time_lib import options_app_now_str
+
+                params.append(options_app_now_str())
                 params.append(int(row["id"]))
                 conn.execute(
                     f"""
                     UPDATE options_trades
                     SET status = 'closed', close_quote = ?, premium_received = ?,
-                        realized_pnl = ?, close_ord_id = ?, closed_at = CURRENT_TIMESTAMP
+                        realized_pnl = ?, close_ord_id = ?, closed_at = ?
                         {note_sql}
                     WHERE id = ?
                     """,

lib/options/options_coin_open_lib.py

@@ -407,6 +407,8 @@ def open_coin_option_buy_full(
 
 
 def _insert_coin_trade(conn: Any, **kwargs: Any) -> int:
+    from lib.options.options_time_lib import options_app_now_str
+
     pe = 1 if kwargs.get("profit_exit_enabled") else 0
     pe_mult = float(kwargs.get("profit_exit_mult") or 1.0)
     pe_state = "active" if pe else "idle"
@@ -416,8 +418,8 @@ def _insert_coin_trade(conn: Any, **kwargs: Any) -> int:
             inst_id, underlying, opt_type, strike, exp_time, sheets, eth_amount,
             open_quote, premium_paid, status, signal_note, exchange_ord_id,
             margin_mode, premium_ccy, bridge_id, budget_usdt,
-            profit_exit_enabled, profit_exit_mult, profit_exit_state
+            profit_exit_enabled, profit_exit_mult, profit_exit_state, created_at
-        ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 'open', ?, ?, 'coin', ?, ?, ?, ?, ?, ?)
+        ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 'open', ?, ?, 'coin', ?, ?, ?, ?, ?, ?, ?)
         """,
         (
             kwargs["inst_id"],
@@ -437,6 +439,7 @@ def _insert_coin_trade(conn: Any, **kwargs: Any) -> int:
             pe,
             pe_mult,
             pe_state,
+            options_app_now_str(),
         ),
     )
     conn.commit()

lib/options/options_monitor_lib.py

@@ -1,16 +1,12 @@
 """期权持仓监控:浮盈翻倍微信提醒 + 平仓/到期状态同步."""
 from __future__ import annotations
 
-import os
 import sqlite3
 import time
-from datetime import datetime, timezone
 from typing import Any, Callable
-from zoneinfo import ZoneInfo
 
 from lib.exchange.okx_options_lib import normalize_option_exp_ms, resolve_option_close_from_history
-
+from lib.options.options_time_lib import ms_to_options_app_str, options_app_now_str, options_wall_clock_to_ms
-_APP_TZ = ZoneInfo((os.getenv("APP_TIMEZONE") or os.getenv("TZ") or "Asia/Shanghai").strip() or "Asia/Shanghai")
 
 
 def _safe_float(v: Any) -> float | None:
@@ -133,18 +129,7 @@ def run_options_profit_alerts(
 
 def _created_at_ms(created_at: Any) -> int | None:
     """墙钟 created_at → UTC ms.库内时间为业务时区(默认 Asia/Shanghai),不可当 UTC."""
-    if not created_at:
+    return options_wall_clock_to_ms(created_at)
-        return None
-    raw = str(created_at).strip()
-    if not raw:
-        return None
-    for fmt, ln in (("%Y-%m-%d %H:%M:%S", 19), ("%Y-%m-%d %H:%M:%f", 26), ("%Y-%m-%d %H:%M", 16)):
-        try:
-            dt = datetime.strptime(raw[:ln], fmt).replace(tzinfo=_APP_TZ)
-            return int(dt.timestamp() * 1000)
-        except ValueError:
-            continue
-    return None
 
 
 def _group_key_for_closed_trade(row: Any) -> str:
@@ -309,21 +294,20 @@ def sync_open_options_trades(
                 prem_recv = round(paid + float(realized_pnl), 4)
             close_ms = close_info.get("close_ms")
             if close_ms:
-                closed_at = datetime.fromtimestamp(int(close_ms) / 1000, tz=timezone.utc).strftime(
+                closed_at = ms_to_options_app_str(close_ms)
-                    "%Y-%m-%d %H:%M:%S"
-                )
         elif exp_ms is not None and now_ms >= int(exp_ms):
             close_reason = "expired"
             close_quote = 0.0
             prem_recv = 0.0
             realized_pnl = round(-paid, 4)
             if exp_ms:
-                closed_at = datetime.fromtimestamp(int(exp_ms) / 1000, tz=timezone.utc).strftime(
+                closed_at = ms_to_options_app_str(exp_ms)
-                    "%Y-%m-%d %H:%M:%S"
-                )
         else:
             continue
 
+        if not closed_at:
+            closed_at = options_app_now_str()
+
         conn.execute(
             """
             UPDATE options_trades
@@ -332,7 +316,7 @@ def sync_open_options_trades(
                 premium_received = ?,
                 realized_pnl = ?,
                 close_ord_id = COALESCE(?, close_ord_id),
-                closed_at = COALESCE(?, closed_at, CURRENT_TIMESTAMP),
+                closed_at = COALESCE(?, closed_at),
                 signal_note = CASE
                     WHEN ? = 'expired' AND (signal_note IS NULL OR TRIM(signal_note) = '')
                     THEN '到期结算'

lib/options/options_register.py

@@ -1218,6 +1218,7 @@ def register_options_routes(app: Flask, cfg: dict[str, Any]) -> None:
         try:
             init_options_tables(conn)
             from lib.options.options_profit_exit_lib import ensure_profit_exit_columns
+            from lib.options.options_time_lib import options_app_now_str
 
             ensure_profit_exit_columns(conn)
             meta = q.get("meta") or {}
@@ -1230,8 +1231,8 @@ def register_options_routes(app: Flask, cfg: dict[str, Any]) -> None:
                 INSERT INTO options_trades
                 (inst_id, underlying, opt_type, strike, exp_time, sheets, eth_amount,
                  open_quote, premium_paid, status, signal_note, exchange_ord_id,
-                 profit_exit_enabled, profit_exit_mult, profit_exit_state)
+                 profit_exit_enabled, profit_exit_mult, profit_exit_state, created_at)
-                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 'open', ?, ?, ?, ?, ?)
+                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 'open', ?, ?, ?, ?, ?, ?)
                 """,
                 (
                     inst_id,
@@ -1248,6 +1249,7 @@ def register_options_routes(app: Flask, cfg: dict[str, Any]) -> None:
                     1 if profit_exit_enabled else 0,
                     profit_exit_mult if profit_exit_enabled else 1.0,
                     "active" if profit_exit_enabled else "idle",
+                    options_app_now_str(),
                 ),
             )
             trade_id = int(cur.lastrowid)
@@ -1348,6 +1350,8 @@ def register_options_routes(app: Flask, cfg: dict[str, Any]) -> None:
                 conn = cfg["get_db"]()
                 try:
                     init_options_tables(conn)
+                    from lib.options.options_time_lib import options_app_now_str
+
                     conn.execute(
                         """
                         UPDATE options_trades
@@ -1356,10 +1360,10 @@ def register_options_routes(app: Flask, cfg: dict[str, Any]) -> None:
                                 WHEN signal_note IS NULL OR TRIM(signal_note) = '' THEN '委托撤销'
                                 ELSE signal_note
                             END,
-                            closed_at = CURRENT_TIMESTAMP
+                            closed_at = ?
                         WHERE inst_id = ? AND exchange_ord_id = ? AND status = 'open'
                         """,
-                        (inst_id, ord_id),
+                        (options_app_now_str(), inst_id, ord_id),
                     )
                     conn.commit()
                 finally:

lib/options/options_review_lib.py

@@ -715,6 +715,9 @@ def ensure_local_review_synced(
     backfill_exchange_pnl: bool = True,
 ) -> dict[str, Any]:
     """列表/统计前轻量刷新本地源;有交易所时先用历史仓位盈亏覆盖本地再导入复盘."""
+    from lib.options.options_time_lib import migrate_options_utc_wall_clock_to_app
+
+    ts_migrate = migrate_options_utc_wall_clock_to_app(conn)
     if backfill_exchange_pnl and ex is not None:
         try:
             from lib.exchange.okx_options_lib import fetch_all_option_positions_history
@@ -735,6 +738,7 @@ def ensure_local_review_synced(
         out["repaired_usdt"] = repair_coin_review_rows_to_usdt(conn, ex=ex)
     except Exception:
         out["repaired_usdt"] = 0
+    out["ts_migrate"] = ts_migrate
     return out
 
 

lib/options/options_time_lib.py

@@ -0,0 +1,121 @@
+"""期权业务时区墙钟:库内统一按 APP_TIMEZONE(默认 Asia/Shanghai)存储,不可当 UTC."""
+
+from __future__ import annotations
+
+import os
+import sqlite3
+from datetime import datetime, timezone
+from typing import Any, Optional
+from zoneinfo import ZoneInfo
+
+_APP_TZ = ZoneInfo(
+    (os.getenv("APP_TIMEZONE") or os.getenv("TZ") or "Asia/Shanghai").strip()
+    or "Asia/Shanghai"
+)
+_MIGRATE_FLAG = "options_ts_utc_wall_to_app_done"
+
+
+def options_app_tz() -> ZoneInfo:
+    return _APP_TZ
+
+
+def options_app_now_str() -> str:
+    return datetime.now(_APP_TZ).strftime("%Y-%m-%d %H:%M:%S")
+
+
+def ms_to_options_app_str(ms: Any) -> Optional[str]:
+    """交易所 ms(UTC 绝对时间) → 业务时区墙钟字符串."""
+    try:
+        val = float(ms)
+    except (TypeError, ValueError):
+        return None
+    if val <= 0:
+        return None
+    try:
+        dt = datetime.fromtimestamp(int(val) / 1000.0, tz=timezone.utc).astimezone(_APP_TZ)
+        return dt.strftime("%Y-%m-%d %H:%M:%S")
+    except (TypeError, ValueError, OSError, OverflowError):
+        return None
+
+
+def options_wall_clock_to_ms(raw: Any) -> Optional[int]:
+    """业务时区墙钟 → UTC ms(与合约复盘 _local_input_datetime_to_ms 同口径)."""
+    if raw is None or raw == "":
+        return None
+    s = str(raw).strip().replace("T", " ")
+    for fmt, ln in (("%Y-%m-%d %H:%M:%S", 19), ("%Y-%m-%d %H:%M", 16)):
+        try:
+            dt = datetime.strptime(s[:ln], fmt).replace(tzinfo=_APP_TZ)
+            return int(dt.timestamp() * 1000)
+        except ValueError:
+            continue
+    return None
+
+
+def utc_wall_clock_to_app_str(raw: Any) -> Optional[str]:
+    """把误存成 UTC 墙钟的字符串改写为业务时区墙钟(+8 等)."""
+    if raw is None or raw == "":
+        return None
+    s = str(raw).strip().replace("T", " ")
+    for fmt, ln in (("%Y-%m-%d %H:%M:%S", 19), ("%Y-%m-%d %H:%M", 16)):
+        try:
+            dt = datetime.strptime(s[:ln], fmt).replace(tzinfo=timezone.utc)
+            return dt.astimezone(_APP_TZ).strftime("%Y-%m-%d %H:%M:%S")
+        except ValueError:
+            continue
+    return None
+
+
+def migrate_options_utc_wall_clock_to_app(conn: sqlite3.Connection) -> dict[str, int]:
+    """一次性:历史 UTC 墙钟(SQLite CURRENT_TIMESTAMP / 旧 monitor)改为业务时区.
+
+    幂等:同步状态 options_ts_utc_wall_to_app_done=1 后跳过.
+    """
+    from lib.options.options_review_db import init_options_review_tables
+    from lib.options.options_review_lib import get_sync_state, set_sync_state
+
+    init_options_review_tables(conn)
+    if get_sync_state(conn, _MIGRATE_FLAG) == "1":
+        return {"skipped": 1, "trades": 0, "review": 0, "hidden": 0}
+
+    def _shift_table(table: str, cols: tuple[str, ...]) -> int:
+        try:
+            rows = conn.execute(
+                f"SELECT rowid AS _rid, {', '.join(cols)} FROM {table}"
+            ).fetchall()
+        except sqlite3.Error:
+            return 0
+        n = 0
+        for row in rows:
+            sets: list[str] = []
+            vals: list[Any] = []
+            for col in cols:
+                old = row[col]
+                new = utc_wall_clock_to_app_str(old)
+                if new and str(old or "").strip() != new:
+                    sets.append(f"{col}=?")
+                    vals.append(new)
+            if not sets:
+                continue
+            vals.append(int(row["_rid"]))
+            conn.execute(
+                f"UPDATE {table} SET {', '.join(sets)} WHERE rowid=?",
+                vals,
+            )
+            n += 1
+        return n
+
+    trades = 0
+    try:
+        from lib.options.options_db import init_options_tables
+
+        init_options_tables(conn)
+        trades = _shift_table("options_trades", ("created_at", "closed_at"))
+    except Exception:
+        trades = 0
+
+    review = _shift_table("options_review_trades", ("opened_at", "closed_at"))
+    hidden = _shift_table("options_review_hidden", ("closed_at",))
+
+    set_sync_state(conn, _MIGRATE_FLAG, "1")
+    return {"skipped": 0, "trades": trades, "review": review, "hidden": hidden}

tests/test_options_time_lib.py

@@ -0,0 +1,92 @@
+"""期权业务时区与 UTC 墙钟迁移单测."""
+from __future__ import annotations
+
+import sqlite3
+import unittest
+from datetime import datetime, timezone
+from zoneinfo import ZoneInfo
+
+from lib.options.options_db import init_options_tables
+from lib.options.options_review_db import init_options_review_tables
+from lib.options.options_review_lib import get_sync_state, upsert_option_history_row
+from lib.options.options_time_lib import (
+    migrate_options_utc_wall_clock_to_app,
+    ms_to_options_app_str,
+    options_wall_clock_to_ms,
+    utc_wall_clock_to_app_str,
+)
+
+
+class OptionsTimeLibTests(unittest.TestCase):
+    def test_ms_to_shanghai_wall_clock(self):
+        # 2026-09-13 21:41:57 UTC → 2026-09-14 05:41:57 Asia/Shanghai
+        ms = int(
+            datetime(2026, 9, 13, 21, 41, 57, tzinfo=timezone.utc).timestamp() * 1000
+        )
+        self.assertEqual(ms_to_options_app_str(ms), "2026-09-14 05:41:57")
+
+    def test_utc_wall_to_app(self):
+        self.assertEqual(
+            utc_wall_clock_to_app_str("2026-09-13 21:41:57"),
+            "2026-09-14 05:41:57",
+        )
+
+    def test_wall_clock_to_ms_roundtrip(self):
+        raw = "2026-09-14 05:41:57"
+        ms = options_wall_clock_to_ms(raw)
+        self.assertIsNotNone(ms)
+        dt = datetime.fromtimestamp(ms / 1000.0, tz=ZoneInfo("Asia/Shanghai"))
+        self.assertEqual(dt.strftime("%Y-%m-%d %H:%M:%S"), raw)
+
+    def test_migrate_once(self):
+        conn = sqlite3.connect(":memory:")
+        conn.row_factory = sqlite3.Row
+        init_options_tables(conn)
+        init_options_review_tables(conn)
+        conn.execute(
+            """
+            INSERT INTO options_trades(
+                inst_id, underlying, opt_type, sheets, eth_amount, premium_paid, status,
+                created_at, closed_at
+            ) VALUES ('ETH-USD-260917-2525-P', 'ETH', 'P', 1, 0.01, 1, 'closed',
+                      '2026-09-13 21:41:57', '2026-09-14 02:31:00')
+            """
+        )
+        upsert_option_history_row(
+            conn,
+            {
+                "history_key": "local_opt:1",
+                "inst_id": "ETH-USD-260917-2525-P",
+                "underlying": "ETH",
+                "opt_type": "P",
+                "realized_pnl": 0.5,
+                "created_at": "2026-09-13 21:41:57",
+                "closed_at": "2026-09-14 02:31:00",
+            },
+        )
+        conn.commit()
+
+        first = migrate_options_utc_wall_clock_to_app(conn)
+        self.assertEqual(first["skipped"], 0)
+        self.assertGreaterEqual(first["trades"], 1)
+        self.assertGreaterEqual(first["review"], 1)
+
+        row = conn.execute(
+            "SELECT created_at, closed_at FROM options_trades WHERE id=1"
+        ).fetchone()
+        self.assertEqual(row["created_at"], "2026-09-14 05:41:57")
+        self.assertEqual(row["closed_at"], "2026-09-14 10:31:00")
+
+        rev = conn.execute(
+            "SELECT opened_at, closed_at FROM options_review_trades WHERE history_key='local_opt:1'"
+        ).fetchone()
+        self.assertEqual(rev["opened_at"], "2026-09-14 05:41:57")
+        self.assertEqual(rev["closed_at"], "2026-09-14 10:31:00")
+        self.assertEqual(get_sync_state(conn, "options_ts_utc_wall_to_app_done"), "1")
+
+        second = migrate_options_utc_wall_clock_to_app(conn)
+        self.assertEqual(second["skipped"], 1)
+
+
+if __name__ == "__main__":
+    unittest.main()