alphax/app/db/recommendation_queries.py

"""Recommendation and lifecycle-facing DB API."""

from datetime import datetime, timedelta

from app.db.altcoin_db import (
    PUSH_COOLDOWN_HOURS,
    _classify_recommendation_result,
    _derive_execution_fields,
    _is_actionable_execution_status,
    apply_recommendation_state_transition,
    update_recommendation_tracking,
)
from app.db.schema import get_conn


def should_push(symbol: str, push_type: str, action_status: str = "") -> bool:
    """状态感知冷却判断。"""
    conn = get_conn()
    cutoff = (datetime.now() - timedelta(hours=PUSH_COOLDOWN_HOURS)).isoformat()
    if action_status:
        row = conn.execute(
            "SELECT action_status FROM push_log WHERE symbol=? AND push_type=? AND pushed_at > ? ORDER BY id DESC LIMIT 1",
            (symbol, push_type, cutoff),
        ).fetchone()
        conn.close()
        if row is None:
            return True
        return row[0] != action_status

    row = conn.execute(
        "SELECT id FROM push_log WHERE symbol=? AND push_type=? AND pushed_at > ? ORDER BY id DESC LIMIT 1",
        (symbol, push_type, cutoff),
    ).fetchone()
    conn.close()
    return row is None


def log_push(symbol: str, push_type: str, action_status: str = "", rec_id: int = 0):
    """记录一次推送，保留推荐来源可追溯性。"""
    conn = get_conn()
    try:
        cols = [row[1] for row in conn.execute("PRAGMA table_info(push_log)").fetchall()]
        if "rec_id" in cols:
            conn.execute(
                "INSERT INTO push_log (symbol, push_type, action_status, rec_id, pushed_at) VALUES (?,?,?,?,?)",
                (symbol, push_type, action_status, int(rec_id or 0), datetime.now().isoformat()),
            )
        else:
            conn.execute(
                "INSERT INTO push_log (symbol, push_type, action_status, pushed_at) VALUES (?,?,?,?)",
                (symbol, push_type, action_status, datetime.now().isoformat()),
            )
        conn.commit()
    finally:
        conn.close()


def get_recommendation_for_push(rec_id: int):
    """读取单条推荐并派生网站同口径展示状态，供推送层消费。"""
    try:
        rec_id = int(rec_id or 0)
    except Exception:
        rec_id = 0
    if rec_id <= 0:
        return None

    conn = get_conn()
    row = conn.execute(
        """
        SELECT r.*,
               lpc.price AS latest_cache_price,
               lpc.updated_at AS latest_cache_updated_at
        FROM recommendation r
        LEFT JOIN latest_price_cache lpc ON lpc.symbol = r.symbol
        WHERE r.id=?
        """,
        (rec_id,),
    ).fetchone()
    conn.close()
    if not row:
        return None

    item = dict(row)
    rec_result, rec_result_label = _classify_recommendation_result(item)
    item["recommendation_result"] = rec_result
    item["recommendation_result_label"] = rec_result_label
    return _derive_execution_fields(item)


def get_active_recommendations(actionable_only: bool = False):
    """获取所有 active 推荐。"""
    conn = get_conn()
    rows = conn.execute(
        """
        SELECT * FROM recommendation
        WHERE status='active' AND COALESCE(display_bucket,'watch_pool') != 'history'
        ORDER BY rec_time DESC
        """
    ).fetchall()
    conn.close()

    result = []
    for row in rows:
        item = _derive_execution_fields(dict(row))
        if actionable_only and not _is_actionable_execution_status(item.get("execution_status")):
            continue
        result.append(item)
    return result


def get_active_recommendations_deduped(
    actionable_only: bool = True,
    version: str = "",
    hours: float = 0,
    watch_symbols=None,
    limit: int = 0,
    offset: int = 0,
    with_meta: bool = False,
):
    """同 symbol 只保留最新 active 推荐，并附带派生执行状态。"""
    conn = get_conn()
    where = "status='active' AND COALESCE(display_bucket,'watch_pool') != 'history'"
    params = []
    version = str(version or "").strip()
    if version:
        where += " AND strategy_version=?"
        params.append(version)
    if watch_symbols:
        symbols = [str(s).strip().upper() for s in watch_symbols if str(s).strip()]
        if symbols:
            where += " AND symbol IN (" + ",".join(["?"] * len(symbols)) + ")"
            params.extend(symbols)
    try:
        hours = float(hours or 0)
    except Exception:
        hours = 0
    if hours > 0:
        where += " AND julianday(?) - julianday(rec_time) <= ?"
        params.extend([datetime.now().isoformat(), hours / 24.0])

    try:
        limit = max(0, int(limit or 0))
    except Exception:
        limit = 0
    try:
        offset = max(0, int(offset or 0))
    except Exception:
        offset = 0

    rows = conn.execute(
        f"""
        SELECT r.*,
               lpc.price AS latest_cache_price,
               lpc.updated_at AS latest_cache_updated_at
        FROM recommendation r
        LEFT JOIN latest_price_cache lpc ON lpc.symbol = r.symbol
        JOIN (
            SELECT symbol, MAX(id) AS max_id
            FROM recommendation
            WHERE {where}
            GROUP BY symbol
        ) latest ON latest.max_id = r.id
        ORDER BY r.rec_time DESC
        """,
        tuple(params),
    ).fetchall()
    conn.close()

    all_items = []
    summary = {"buy_now": 0, "wait_pullback": 0, "observe": 0, "observe_strong": 0, "observe_weak": 0, "expired": 0, "total": 0}
    now = datetime.now()
    for row in rows:
        item = dict(row)
        rec_result, rec_result_label = _classify_recommendation_result(item)
        item["recommendation_result"] = rec_result
        item["recommendation_result_label"] = rec_result_label
        _derive_execution_fields(item)

        is_expired = False
        if hours > 0:
            try:
                rec_time = item.get("rec_time")
                if rec_time:
                    is_expired = (now - datetime.fromisoformat(str(rec_time))).total_seconds() > hours * 3600
            except Exception:
                is_expired = False
            if item.get("execution_status") == "invalid" or item.get("status") in ("invalid", "expired", "archived") or item.get("display_bucket") == "history":
                is_expired = True
        if is_expired:
            summary["expired"] += 1
            continue

        if actionable_only and not _is_actionable_execution_status(item.get("execution_status")):
            continue
        all_items.append(item)
        if item.get("execution_status") == "buy_now":
            summary["buy_now"] += 1
        elif item.get("execution_status") == "wait_pullback":
            summary["wait_pullback"] += 1
        else:
            summary["observe"] += 1
            if item.get("observe_tier") == "weak":
                summary["observe_weak"] += 1
            else:
                summary["observe_strong"] += 1

    summary["total"] = len(all_items)
    summary["expired_filtered"] = summary.pop("expired", 0)

    if not with_meta:
        return all_items
    page_items = all_items[offset : offset + limit] if limit else all_items[offset:]
    return {
        "items": page_items,
        "total": len(all_items),
        "limit": limit,
        "offset": offset,
        "has_more": bool(limit and offset + len(page_items) < len(all_items)),
        "summary": summary,
    }

__all__ = [
    "apply_recommendation_state_transition",
    "get_active_recommendations",
    "get_active_recommendations_deduped",
    "get_recommendation_for_push",
    "log_push",
    "should_push",
    "update_recommendation_tracking",
]