dekun
192 lines
6.7 KiB
Python
192 lines
6.7 KiB
Python
"""期权业务时区墙钟:库内统一按 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"
|
|
_SWAP_REPAIR_FLAG = "options_closed_at_note_swap_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 sqlite3.OperationalError:
|
|
return {"skipped": 0, "trades": 0, "review": 0, "hidden": 0, "locked": 1}
|
|
except Exception:
|
|
trades = 0
|
|
|
|
try:
|
|
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")
|
|
return {"skipped": 0, "trades": trades, "review": review, "hidden": hidden}
|
|
|
|
|
|
def repair_swapped_closed_at_signal_note(conn: sqlite3.Connection) -> dict[str, int]:
|
|
"""修复 close_option_by_bid1 曾把 signal_note 写入 closed_at 的坏行.
|
|
|
|
典型表现: closed_at='翻倍出场'/'目标位平仓', signal_note 反而是时间戳.
|
|
幂等:同步状态 options_closed_at_note_swap_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)
|
|
trades_fixed = 0
|
|
review_fixed = 0
|
|
try:
|
|
from lib.options.options_db import init_options_tables
|
|
|
|
init_options_tables(conn)
|
|
rows = conn.execute(
|
|
"""
|
|
SELECT id, closed_at, signal_note
|
|
FROM options_trades
|
|
WHERE status = 'closed'
|
|
AND closed_at IS NOT NULL
|
|
AND TRIM(closed_at) != ''
|
|
"""
|
|
).fetchall()
|
|
for row in rows:
|
|
closed = str(row["closed_at"] or "").strip()
|
|
note = str(row["signal_note"] or "").strip()
|
|
if options_wall_clock_to_ms(closed) is not None:
|
|
continue
|
|
if options_wall_clock_to_ms(note) is None:
|
|
continue
|
|
conn.execute(
|
|
"""
|
|
UPDATE options_trades
|
|
SET closed_at = ?, signal_note = ?
|
|
WHERE id = ?
|
|
""",
|
|
(note, closed, int(row["id"])),
|
|
)
|
|
trades_fixed += 1
|
|
hk = f"local_opt:{int(row['id'])}"
|
|
cur = conn.execute(
|
|
"""
|
|
UPDATE options_review_trades
|
|
SET closed_at = ?
|
|
WHERE history_key = ?
|
|
AND (closed_at IS NULL OR TRIM(closed_at) = '' OR closed_at = ?)
|
|
""",
|
|
(note, hk, closed),
|
|
)
|
|
review_fixed += int(cur.rowcount or 0)
|
|
except sqlite3.OperationalError:
|
|
# 并发刷新撞锁时跳过本轮修复,不影响列表读取
|
|
return {"trades": 0, "review": 0, "locked": 1}
|
|
except Exception:
|
|
pass
|
|
|
|
# 已标记过则不再写同步状态,减少并发写冲突
|
|
if get_sync_state(conn, _SWAP_REPAIR_FLAG) != "1":
|
|
set_sync_state(conn, _SWAP_REPAIR_FLAG, "1")
|
|
return {"trades": trades_fixed, "review": review_fixed}
|