diff --git a/lib/exchange/okx_options_lib.py b/lib/exchange/okx_options_lib.py index 36d9051..6b84c30 100644 --- a/lib/exchange/okx_options_lib.py +++ b/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) - if val is None or val <= 0: - return None - try: - from datetime import datetime, timezone + """OKX ms → 业务时区墙钟(默认 Asia/Shanghai);勿用系统本地时区.""" + from lib.options.options_time_lib import ms_to_options_app_str - dt = datetime.fromtimestamp(int(val) / 1000.0, tz=timezone.utc).astimezone() - return dt.strftime("%Y-%m-%d %H:%M:%S") - except (TypeError, ValueError, OSError): - return None + return ms_to_options_app_str(ms) def option_instrument_meta_cached( diff --git a/lib/options/options_close_exec_lib.py b/lib/options/options_close_exec_lib.py index 07171e3..acde02a 100644 --- a/lib/options/options_close_exec_lib.py +++ b/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 = ? """, diff --git a/lib/options/options_coin_open_lib.py b/lib/options/options_coin_open_lib.py index eeb2e48..6237b41 100644 --- a/lib/options/options_coin_open_lib.py +++ b/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 - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 'open', ?, ?, 'coin', ?, ?, ?, ?, ?, ?) + profit_exit_enabled, profit_exit_mult, profit_exit_state, created_at + ) 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() diff --git a/lib/options/options_monitor_lib.py b/lib/options/options_monitor_lib.py index f2a6194..64bbdf5 100644 --- a/lib/options/options_monitor_lib.py +++ b/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 - -_APP_TZ = ZoneInfo((os.getenv("APP_TIMEZONE") or os.getenv("TZ") or "Asia/Shanghai").strip() or "Asia/Shanghai") +from lib.options.options_time_lib import ms_to_options_app_str, options_app_now_str, options_wall_clock_to_ms 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 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 + return options_wall_clock_to_ms(created_at) 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( - "%Y-%m-%d %H:%M:%S" - ) + closed_at = ms_to_options_app_str(close_ms) 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( - "%Y-%m-%d %H:%M:%S" - ) + closed_at = ms_to_options_app_str(exp_ms) 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 '到期结算' diff --git a/lib/options/options_register.py b/lib/options/options_register.py index 9ca4579..6f3c54f 100644 --- a/lib/options/options_register.py +++ b/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) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 'open', ?, ?, ?, ?, ?) + profit_exit_enabled, profit_exit_mult, profit_exit_state, created_at) + 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: diff --git a/lib/options/options_review_lib.py b/lib/options/options_review_lib.py index 5fff340..6dc0733 100644 --- a/lib/options/options_review_lib.py +++ b/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 diff --git a/lib/options/options_time_lib.py b/lib/options/options_time_lib.py new file mode 100644 index 0000000..83b2fcb --- /dev/null +++ b/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} diff --git a/tests/test_options_time_lib.py b/tests/test_options_time_lib.py new file mode 100644 index 0000000..6b26944 --- /dev/null +++ b/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()