buffer_platform/buffer_platform_v2/app/services/cache.py

"""缓存服务模块 - 数据缓存检查、保存、查询以及任务管理（三层缓存：内存→Redis→数据库）"""

import json
import logging
from datetime import datetime
from typing import Any

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.config import settings
from app.models.market import MarketData, ScheduledTask, SymbolTimestamp
from app.services.redis_cache import redis_cache

logger = logging.getLogger(__name__)


# ==================== 市场数据缓存 ====================


async def is_cache_valid(
    db: AsyncSession,
    symbol: str,
    data_type: str,
    period: str,
    ttl_seconds: int = settings.cache_ttl_seconds,
) -> bool:
    """判断指定品种+周期缓存是否在 TTL 有效期内。"""
    stmt = select(MarketData).where(
        MarketData.symbol == symbol,
        MarketData.data_type == data_type,
        MarketData.period == period,
    )
    result = await db.execute(stmt)
    record = result.scalar_one_or_none()
    if not record:
        return False
    age = (datetime.now() - record.fetched_at).total_seconds()
    return age < ttl_seconds


async def check_cache_status(
    db: AsyncSession,
    symbol: str,
    data_type: str,
    periods: list[str],
    ttl_seconds: int = settings.cache_ttl_seconds,
) -> dict[str, Any]:
    """检查多个周期的缓存状态，返回 all_valid / missing_periods / valid_periods。"""
    valid: list[str] = []
    missing: list[str] = []
    for p in periods:
        if await is_cache_valid(db, symbol, data_type, p, ttl_seconds):
            valid.append(p)
        else:
            missing.append(p)
    return {
        "all_valid": len(missing) == 0,
        "valid_periods": valid,
        "missing_periods": missing,
    }


async def save_market_data(
    db: AsyncSession,
    symbol: str,
    data: dict[str, Any],
) -> MarketData:
    """将采集到的市场数据保存到缓存表。"""
    now = datetime.now()

    for period, candles in data.get("timeframes", {}).items():
        stmt = select(MarketData).where(
            MarketData.symbol == symbol,
            MarketData.data_type == data.get("type", "futures"),
            MarketData.period == period,
        )
        result = await db.execute(stmt)
        record = result.scalar_one_or_none()

        candles_json = json.dumps(candles, ensure_ascii=False)

        if record:
            record.candles_json = candles_json
            record.current_price = data.get("current_price")
            record.fetched_at = now
            record.candle_count = len(candles)
        else:
            record = MarketData(
                symbol=symbol,
                data_type=data.get("type", "futures"),
                period=period,
                candles_json=candles_json,
                current_price=data.get("current_price"),
                fetched_at=now,
                candle_count=len(candles),
            )
            db.add(record)

    await update_symbol_timestamp(db, symbol, data.get("type", "futures"), now)
    await db.flush()
    logger.info("缓存已更新: %s, %d 个周期", symbol, len(data.get("timeframes", {})))

    # 回填 Redis 缓存
    for period, candles in data.get("timeframes", {}).items():
        await redis_cache.set_kline(symbol, period, {
            "candles": candles,
            "current_price": data.get("current_price"),
            "fetched_at": now.isoformat(),
        })

    stmt = (
        select(MarketData)
        .where(
            MarketData.symbol == symbol,
            MarketData.data_type == data.get("type", "futures"),
        )
        .order_by(MarketData.fetched_at.desc())
    )
    result = await db.execute(stmt)
    return result.scalars().first()


async def update_symbol_timestamp(
    db: AsyncSession,
    symbol: str,
    data_type: str,
    refresh_time: datetime,
) -> None:
    """更新品种数据时间戳记录。"""
    stmt = select(SymbolTimestamp).where(
        SymbolTimestamp.symbol == symbol,
        SymbolTimestamp.data_type == data_type,
    )
    result = await db.execute(stmt)
    record = result.scalar_one_or_none()

    if record:
        record.last_refresh_at = refresh_time
        record.refresh_count += 1
    else:
        record = SymbolTimestamp(
            symbol=symbol,
            data_type=data_type,
            last_refresh_at=refresh_time,
            refresh_count=1,
        )
        db.add(record)

    await db.flush()


async def get_symbol_timestamp(
    db: AsyncSession,
    symbol: str,
    data_type: str = "futures",
) -> datetime | None:
    """获取品种最新数据时间戳。"""
    stmt = select(SymbolTimestamp).where(
        SymbolTimestamp.symbol == symbol,
        SymbolTimestamp.data_type == data_type,
    )
    result = await db.execute(stmt)
    record = result.scalar_one_or_none()
    return record.last_refresh_at if record else None


async def needs_refresh(
    db: AsyncSession,
    symbol: str,
    data_type: str = "futures",
    threshold_seconds: int = 300,
) -> bool:
    """判断品种数据是否需要刷新。"""
    last_refresh = await get_symbol_timestamp(db, symbol, data_type)
    if last_refresh is None:
        return True
    age = (datetime.now() - last_refresh).total_seconds()
    return age > threshold_seconds


async def get_latest_cached(
    db: AsyncSession,
    symbol: str,
    data_type: str = "futures",
    period: str | None = None,
) -> list[MarketData]:
    """获取品种最新缓存的 MarketData 记录列表。"""
    stmt = select(MarketData).where(
        MarketData.symbol == symbol,
        MarketData.data_type == data_type,
    )
    if period:
        stmt = stmt.where(MarketData.period == period)
    stmt = stmt.order_by(MarketData.fetched_at.desc())
    result = await db.execute(stmt)
    return list(result.scalars().all())


async def get_cached_data(
    db: AsyncSession,
    symbol: str,
    data_type: str = "futures",
    periods: list[str] | None = None,
    end_time: datetime | None = None,
    max_candles: int = 100,
) -> dict[str, Any] | None:
    """获取品种的缓存数据（三层缓存：内存→Redis→数据库）。

    优先从 Redis 获取，缺失的周期再从数据库查询并回填 Redis。
    """
    now = datetime.now()
    filter_end_time = end_time if end_time else now
    if filter_end_time.tzinfo is not None:
        filter_end_time = filter_end_time.replace(tzinfo=None)

    # 如果未指定 periods，直接走数据库查询
    if not periods:
        return await _query_from_db_full(db, symbol, data_type, filter_end_time, max_candles, now)

    # L2: 尝试从 Redis 获取各周期数据
    result_timeframes: dict[str, list[Any]] = {}
    missing_periods: list[str] = []
    current_price: float | None = None
    newest_ts: datetime | None = None

    for period in periods:
        cached = await redis_cache.get_kline(symbol, period)
        if cached and "candles" in cached:
            candles = cached["candles"]
            candles = _filter_candles(candles, filter_end_time, max_candles)
            result_timeframes[period] = candles
            if cached.get("current_price") is not None and current_price is None:
                current_price = cached["current_price"]
            if cached.get("fetched_at"):
                try:
                    ts = datetime.fromisoformat(cached["fetched_at"])
                    if newest_ts is None or ts > newest_ts:
                        newest_ts = ts
                except ValueError:
                    pass
        else:
            missing_periods.append(period)

    # L3: 从数据库查询缺失的周期
    if missing_periods:
        db_result = await _query_from_db_periods(
            db, symbol, data_type, missing_periods, filter_end_time, max_candles,
        )
        if db_result:
            result_timeframes.update(db_result["timeframes"])
            if current_price is None and db_result.get("current_price") is not None:
                current_price = db_result["current_price"]
            if db_result.get("fetched_at"):
                try:
                    ts = datetime.fromisoformat(db_result["fetched_at"])
                    if newest_ts is None or ts > newest_ts:
                        newest_ts = ts
                except ValueError:
                    pass

    if not result_timeframes:
        return None

    is_fresh = newest_ts is not None and (now - newest_ts).total_seconds() < settings.cache_ttl_seconds

    return {
        "symbol": symbol,
        "type": data_type,
        "current_price": current_price,
        "timestamp": newest_ts.isoformat() if newest_ts else now.isoformat(),
        "timeframes": result_timeframes,
        "is_fresh": is_fresh,
        "fetched_at": newest_ts.isoformat() if newest_ts else now.isoformat(),
    }


def _filter_candles(
    candles: list[Any],
    filter_end_time: datetime,
    max_candles: int,
) -> list[Any]:
    """过滤K线数据：按截止时间 + 最大条数。"""
    filtered: list[Any] = []
    for candle in candles:
        candle_time = candle.get("datetime") or candle.get("time")
        if candle_time:
            if isinstance(candle_time, str):
                try:
                    candle_dt = datetime.fromisoformat(candle_time.replace("Z", "+00:00"))
                    if candle_dt.tzinfo is not None:
                        candle_dt = candle_dt.replace(tzinfo=None)
                except ValueError:
                    filtered.append(candle)
                    continue
            else:
                candle_dt = candle_time
                if candle_dt.tzinfo is not None:
                    candle_dt = candle_dt.replace(tzinfo=None)
            if candle_dt <= filter_end_time:
                filtered.append(candle)
        else:
            filtered.append(candle)

    if len(filtered) > max_candles:
        filtered = filtered[-max_candles:]
    return filtered


async def _query_from_db_periods(
    db: AsyncSession,
    symbol: str,
    data_type: str,
    periods: list[str],
    filter_end_time: datetime,
    max_candles: int,
) -> dict[str, Any] | None:
    """从数据库查询指定周期的数据，并回填 Redis。"""
    stmt = select(MarketData).where(
        MarketData.symbol == symbol,
        MarketData.data_type == data_type,
        MarketData.period.in_(periods),
    )
    result = await db.execute(stmt)
    records = list(result.scalars().all())
    if not records:
        return None

    timeframes: dict[str, list[Any]] = {}
    current_price: float | None = None
    newest: datetime | None = None

    for r in records:
        candles = json.loads(r.candles_json)
        candles = _filter_candles(candles, filter_end_time, max_candles)
        timeframes[r.period] = candles

        # 回填 Redis
        await redis_cache.set_kline(symbol, r.period, {
            "candles": json.loads(r.candles_json),
            "current_price": r.current_price,
            "fetched_at": r.fetched_at.isoformat(),
        })

        if current_price is None:
            current_price = r.current_price
        if newest is None or r.fetched_at > newest:
            newest = r.fetched_at

    return {
        "symbol": symbol,
        "type": data_type,
        "current_price": current_price,
        "timestamp": newest.isoformat() if newest else "",
        "timeframes": timeframes,
        "is_fresh": False,
        "fetched_at": newest.isoformat() if newest else "",
    }


async def _query_from_db_full(
    db: AsyncSession,
    symbol: str,
    data_type: str,
    filter_end_time: datetime,
    max_candles: int,
    now: datetime,
) -> dict[str, Any] | None:
    """从数据库查询品种全部数据（未指定 periods 时的回退路径）。"""
    stmt = select(MarketData).where(
        MarketData.symbol == symbol,
        MarketData.data_type == data_type,
    )
    result = await db.execute(stmt)
    records = list(result.scalars().all())
    if not records:
        return None

    newest = max(r.fetched_at for r in records)
    is_fresh = (now - newest).total_seconds() < settings.cache_ttl_seconds

    timeframes: dict[str, list[Any]] = {}
    current_price: float | None = None
    for r in records:
        candles = json.loads(r.candles_json)
        candles = _filter_candles(candles, filter_end_time, max_candles)
        timeframes[r.period] = candles
        if current_price is None:
            current_price = r.current_price

    return {
        "symbol": symbol,
        "type": data_type,
        "current_price": current_price,
        "timestamp": newest.isoformat(),
        "timeframes": timeframes,
        "is_fresh": is_fresh,
        "fetched_at": newest.isoformat(),
    }


# ==================== 定时任务管理 ====================


async def create_task(
    db: AsyncSession,
    symbol: str,
    data_type: str,
    periods: str,
    interval_seconds: int,
    task_type: str = "interval",
    run_time: str | None = None,
) -> ScheduledTask:
    """创建新的定时采集任务。"""
    stmt = select(ScheduledTask).where(
        ScheduledTask.symbol == symbol,
        ScheduledTask.data_type == data_type,
    )
    result = await db.execute(stmt)
    existing = result.scalar_one_or_none()

    if existing:
        existing.periods = periods
        existing.interval_seconds = interval_seconds
        existing.task_type = task_type
        existing.run_time = run_time
        existing.enabled = True
        existing.updated_at = datetime.now()
        await db.flush()
        return existing

    task = ScheduledTask(
        symbol=symbol,
        data_type=data_type,
        periods=periods,
        interval_seconds=interval_seconds,
        task_type=task_type,
        run_time=run_time,
        enabled=True,
    )
    db.add(task)
    await db.flush()
    return task


async def list_tasks(db: AsyncSession) -> list[ScheduledTask]:
    """列出所有定时任务。"""
    stmt = select(ScheduledTask).order_by(ScheduledTask.created_at.desc())
    result = await db.execute(stmt)
    return list(result.scalars().all())


async def get_task(db: AsyncSession, task_id: int) -> ScheduledTask | None:
    """获取单个任务详情。"""
    stmt = select(ScheduledTask).where(ScheduledTask.id == task_id)
    result = await db.execute(stmt)
    return result.scalar_one_or_none()


async def disable_task(db: AsyncSession, task_id: int) -> ScheduledTask | None:
    """禁用定时任务。"""
    stmt = select(ScheduledTask).where(ScheduledTask.id == task_id)
    result = await db.execute(stmt)
    task = result.scalar_one_or_none()
    if task:
        task.enabled = False
        task.updated_at = datetime.now()
        await db.flush()
    return task


async def enable_task(db: AsyncSession, task_id: int) -> ScheduledTask | None:
    """启用定时任务。"""
    stmt = select(ScheduledTask).where(ScheduledTask.id == task_id)
    result = await db.execute(stmt)
    task = result.scalar_one_or_none()
    if task:
        task.enabled = True
        task.updated_at = datetime.now()
        await db.flush()
    return task


async def delete_task(db: AsyncSession, task_id: int) -> bool:
    """删除定时任务。"""
    stmt = select(ScheduledTask).where(ScheduledTask.id == task_id)
    result = await db.execute(stmt)
    task = result.scalar_one_or_none()
    if task:
        await db.delete(task)
        await db.flush()
        return True
    return False


async def update_task_status(
    db: AsyncSession,
    task_id: int,
    status: str,
    last_run: datetime | None = None,
) -> None:
    """更新任务执行状态。"""
    stmt = select(ScheduledTask).where(ScheduledTask.id == task_id)
    result = await db.execute(stmt)
    task = result.scalar_one_or_none()
    if task:
        task.last_run = last_run or datetime.now()
        task.last_status = status
        await db.flush()