qihuo/modules/market/kline_store.py

# Copyright (c) 2025-2026 马建军. All rights reserved.
# 专有软件 — 未经授权禁止复制、传播、转售。
# 严禁用于：带单/代客理财、向他人推荐期货品种或买卖建议、融资配资等业务。
# 详见 LICENSE.zh-CN.txt 与 docs/软件购买与使用协议.md

"""K 线本地 SQLite 缓存。"""
from __future__ import annotations

import sqlite3
from datetime import datetime, timedelta
from typing import Optional
from zoneinfo import ZoneInfo

TZ = ZoneInfo("Asia/Shanghai")

REFRESH_SECONDS = {
    "timeshare": 30,
    "1m": 30,
    "2m": 30,
    "5m": 60,
    "15m": 60,
    "1h": 120,
    "2h": 120,
    "4h": 180,
    "d": 300,
    "w": 600,
}


def ensure_kline_tables(conn: sqlite3.Connection) -> None:
    conn.execute(
        """CREATE TABLE IF NOT EXISTS kline_bars (
            chart_symbol TEXT NOT NULL,
            period TEXT NOT NULL,
            bar_time TEXT NOT NULL,
            open REAL NOT NULL,
            high REAL NOT NULL,
            low REAL NOT NULL,
            close REAL NOT NULL,
            volume REAL DEFAULT 0,
            updated_at TEXT NOT NULL,
            PRIMARY KEY (chart_symbol, period, bar_time)
        )"""
    )
    conn.execute(
        """CREATE TABLE IF NOT EXISTS kline_meta (
            chart_symbol TEXT NOT NULL,
            period TEXT NOT NULL,
            bar_count INTEGER DEFAULT 0,
            last_bar_time TEXT,
            updated_at TEXT NOT NULL,
            PRIMARY KEY (chart_symbol, period)
        )"""
    )
    conn.execute(
        "CREATE INDEX IF NOT EXISTS idx_kline_bars_sym_period "
        "ON kline_bars(chart_symbol, period, bar_time)"
    )
    conn.commit()


def _parse_updated_at(value: str) -> Optional[datetime]:
    if not value:
        return None
    try:
        return datetime.fromisoformat(value.strip()).replace(tzinfo=TZ)
    except ValueError:
        return None


def is_cache_fresh(period: str, updated_at: str) -> bool:
    dt = _parse_updated_at(updated_at)
    if not dt:
        return False
    ttl = REFRESH_SECONDS.get((period or "").lower(), 60)
    return datetime.now(TZ) - dt < timedelta(seconds=ttl)


def load_bars(conn: sqlite3.Connection, chart_symbol: str, period: str) -> list[dict]:
    rows = conn.execute(
        """SELECT bar_time, open, high, low, close, volume
           FROM kline_bars
           WHERE chart_symbol=? AND period=?
           ORDER BY bar_time ASC""",
        (chart_symbol, period),
    ).fetchall()
    return [
        {
            "d": row[0],
            "o": float(row[1]),
            "h": float(row[2]),
            "l": float(row[3]),
            "c": float(row[4]),
            "v": float(row[5] or 0),
        }
        for row in rows
    ]


def load_meta(conn: sqlite3.Connection, chart_symbol: str, period: str) -> Optional[dict]:
    row = conn.execute(
        "SELECT bar_count, last_bar_time, updated_at FROM kline_meta "
        "WHERE chart_symbol=? AND period=?",
        (chart_symbol, period),
    ).fetchone()
    if not row:
        return None
    return {
        "bar_count": row[0],
        "last_bar_time": row[1],
        "updated_at": row[2],
    }


def save_bars(conn: sqlite3.Connection, chart_symbol: str, period: str, bars: list[dict]) -> int:
    if not bars:
        return 0
    ensure_kline_tables(conn)
    now = datetime.now(TZ).isoformat(timespec="seconds")
    for bar in bars:
        conn.execute(
            """INSERT INTO kline_bars
               (chart_symbol, period, bar_time, open, high, low, close, volume, updated_at)
               VALUES (?,?,?,?,?,?,?,?,?)
               ON CONFLICT(chart_symbol, period, bar_time) DO UPDATE SET
                 open=excluded.open,
                 high=excluded.high,
                 low=excluded.low,
                 close=excluded.close,
                 volume=excluded.volume,
                 updated_at=excluded.updated_at""",
            (
                chart_symbol,
                period,
                str(bar["d"]),
                float(bar["o"]),
                float(bar["h"]),
                float(bar["l"]),
                float(bar["c"]),
                float(bar.get("v") or 0),
                now,
            ),
        )
    last_time = str(bars[-1]["d"])
    conn.execute(
        """INSERT INTO kline_meta (chart_symbol, period, bar_count, last_bar_time, updated_at)
           VALUES (?,?,?,?,?)
           ON CONFLICT(chart_symbol, period) DO UPDATE SET
             bar_count=excluded.bar_count,
             last_bar_time=excluded.last_bar_time,
             updated_at=excluded.updated_at""",
        (chart_symbol, period, len(bars), last_time, now),
    )
    conn.commit()
    return len(bars)


def get_cached_entry(
    conn: sqlite3.Connection,
    chart_symbol: str,
    period: str,
) -> Optional[dict]:
    if not chart_symbol:
        return None
    ensure_kline_tables(conn)
    meta = load_meta(conn, chart_symbol, period)
    bars = load_bars(conn, chart_symbol, period)
    if not bars:
        return None
    updated_at = meta["updated_at"] if meta else ""
    return {
        "bars": bars,
        "updated_at": updated_at,
        "fresh": is_cache_fresh(period, updated_at),
    }