crypto_monitor/strategy_ui.py

"""策略交易页：主站 index.html 所需数据（顺势加仓等）。"""
from __future__ import annotations

from typing import Any, Callable, Optional

from strategy_db import init_strategy_tables
from strategy_roll_monitor_lib import roll_leg_status_label


def _row_to_dict(row) -> dict:
    if row is None:
        return {}
    try:
        return dict(row)
    except Exception:
        return {}


def count_active_trend_plans(conn, count_fn: Optional[Callable] = None) -> int:
    if callable(count_fn):
        return int(count_fn(conn) or 0)
    try:
        return int(
            conn.execute(
                "SELECT COUNT(*) FROM trend_pullback_plans WHERE status='active'"
            ).fetchone()[0]
        )
    except Exception:
        return 0


def fetch_roll_page_data(
    conn,
    *,
    default_risk_percent: float = 2.0,
    count_active_trends: Optional[Callable] = None,
    roll_cfg: dict | None = None,
) -> dict[str, Any]:
    init_strategy_tables(conn)
    monitors = []
    for row in conn.execute(
        "SELECT * FROM order_monitors WHERE status='active' ORDER BY id DESC"
    ).fetchall():
        monitors.append(_row_to_dict(row))
    roll_groups = []
    for row in conn.execute(
        """SELECT g.* FROM roll_groups g
           INNER JOIN order_monitors m ON m.id = g.order_monitor_id AND m.status='active'
           WHERE g.status='active'
           ORDER BY g.id DESC"""
    ).fetchall():
        roll_groups.append(_row_to_dict(row))
    active_gids = {int(g["id"]) for g in roll_groups if g.get("id") is not None}
    roll_legs = []
    for row in conn.execute(
        "SELECT * FROM roll_legs ORDER BY id DESC LIMIT 80"
    ).fetchall():
        leg = _row_to_dict(row)
        gid = leg.get("roll_group_id")
        if gid is not None and int(gid) not in active_gids:
            continue
        leg["status_label"] = roll_leg_status_label(leg.get("status"))
        roll_legs.append(leg)
    roll_legs = roll_legs[:50]
    out = {
        "roll_monitors": monitors,
        "roll_groups": roll_groups,
        "roll_legs": roll_legs,
        "roll_trend_active": count_active_trend_plans(conn, count_active_trends),
        "default_risk_percent": default_risk_percent,
    }
    if roll_cfg:
        from strategy_roll_ui_lib import enrich_roll_page_data

        enrich_roll_page_data(conn, out, roll_cfg)
    return out


DEFAULT_TREND_DISABLED_NOTE = (
    "趋势回调（预览、自动补仓、程序止盈）仅在 Gate 趋势机器人实例 "
    "（crypto_monitor_gate_bot，常见端口 5002）中启用。"
    "币安 / Gate 主站 / OKX 可使用本页「顺势加仓」；完整趋势回调请打开该实例。"
)


def strategy_render_extras(
    conn,
    page: str,
    *,
    default_risk_percent: float = 2.0,
    count_active_trends: Optional[Callable] = None,
    trend_disabled_note: str = "",
    request_obj=None,
    trend_cfg: Optional[dict] = None,
) -> dict[str, Any]:
    """render_main_page 策略相关页变量（含策略交易记录）。"""
    if page == "strategy_records":
        from strategy_records_register import load_strategy_records_page

        return load_strategy_records_page(conn)
    return strategy_page_template_vars(
        conn,
        page,
        default_risk_percent=default_risk_percent,
        count_active_trends=count_active_trends,
        trend_disabled_note=trend_disabled_note,
        request_obj=request_obj,
        trend_cfg=trend_cfg,
    )


def strategy_page_template_vars(
    conn,
    page: str,
    *,
    default_risk_percent: float = 2.0,
    count_active_trends: Optional[Callable] = None,
    trend_disabled_note: str = "",
    request_obj=None,
    trend_cfg: Optional[dict] = None,
) -> dict[str, Any]:
    """render_main_page 在 conn.close() 前合并进 render_template 的变量。"""
    if page not in ("strategy", "strategy_trend", "strategy_roll"):
        return {}
    roll_cfg = None
    try:
        from flask import current_app

        roll_cfg = (current_app.extensions or {}).get("strategy_roll_cfg")
    except Exception:
        roll_cfg = None
    out = fetch_roll_page_data(
        conn,
        default_risk_percent=default_risk_percent,
        count_active_trends=count_active_trends,
        roll_cfg=roll_cfg if isinstance(roll_cfg, dict) else None,
    )
    if trend_cfg and request_obj is not None:
        from strategy_trend_register import load_trend_page_context

        out.update(load_trend_page_context(conn, request_obj, trend_cfg))
    elif page == "strategy_trend":
        out["trend_disabled_note"] = trend_disabled_note or DEFAULT_TREND_DISABLED_NOTE
    return out