fix: show archive chart times in UTC+8 and parse Beijing wall clock

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
dekun
2026-06-08 16:47:09 +08:00
parent 947b58084d
commit 55a979eee5
5 changed files with 103 additions and 32 deletions
+19 -9
View File
@@ -6,9 +6,12 @@ import json
import os
import sqlite3
import time
from datetime import datetime
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Callable, Optional
from zoneinfo import ZoneInfo
CHART_DISPLAY_TZ = ZoneInfo(os.getenv("APP_TIMEZONE", "Asia/Shanghai"))
from hub_ohlcv_lib import (
TIMEFRAME_MS,
@@ -143,7 +146,8 @@ def _now_ms() -> int:
return int(time.time() * 1000)
def _parse_dt_ms(raw: Any) -> int | None:
def parse_wall_clock_ms(raw: Any, *, tz: ZoneInfo = CHART_DISPLAY_TZ) -> int | None:
"""将 YYYY-MM-DD[ HH:MM[:SS]] 按指定时区墙钟解析为 UTC 毫秒(默认 UTC+8)。"""
if raw in (None, ""):
return None
try:
@@ -158,12 +162,22 @@ def _parse_dt_ms(raw: Any) -> int | None:
for fmt, ln in (("%Y-%m-%d %H:%M:%S", 19), ("%Y-%m-%d %H:%M", 16), ("%Y-%m-%d", 10)):
try:
dt = datetime.strptime(s[:ln], fmt)
return int(dt.timestamp() * 1000)
aware = dt.replace(tzinfo=tz)
return int(aware.timestamp() * 1000)
except ValueError:
continue
return None
def ms_to_wall_clock_str(ms: int, *, tz: ZoneInfo = CHART_DISPLAY_TZ) -> str:
dt = datetime.fromtimestamp(int(ms) / 1000.0, tz=timezone.utc).astimezone(tz)
return dt.strftime("%Y-%m-%d %H:%M:%S")
def _parse_dt_ms(raw: Any) -> int | None:
return parse_wall_clock_ms(raw)
def _trade_entry_reason_for_cache(t: dict[str, Any]) -> str:
for key in ("entry_type", "entry_reason", "reviewed_entry_reason"):
raw = t.get(key)
@@ -341,13 +355,9 @@ def _enrich_trade_display_fields(out: dict[str, Any]) -> dict[str, Any]:
if closed_ms:
out["closed_at_ms"] = int(closed_ms)
if not out.get("opened_at") and opened_ms:
out["opened_at"] = datetime.fromtimestamp(int(opened_ms) / 1000).strftime(
"%Y-%m-%d %H:%M:%S"
)
out["opened_at"] = ms_to_wall_clock_str(int(opened_ms))
if not out.get("closed_at") and closed_ms:
out["closed_at"] = datetime.fromtimestamp(int(closed_ms) / 1000).strftime(
"%Y-%m-%d %H:%M:%S"
)
out["closed_at"] = ms_to_wall_clock_str(int(closed_ms))
entry_type = display_entry_type_label(out)
if entry_type and entry_type != "":
out["entry_type"] = entry_type