qihuo/modules/risk/daily_loss_guard.py

# Copyright (c) 2025-2026 马建军. All rights reserved.
"""日亏损风控：达权益比例上限后强制清仓并当日禁开。"""

from __future__ import annotations

import logging
import threading
import time
from typing import Any, Callable, Optional

from modules.core.contract_specs import calc_position_metrics
from modules.market.market_sessions import is_trading_session
from modules.risk.account_risk_lib import (
    _default_get_setting,
    daily_loss_force_close_pct,
    daily_loss_slippage_buffer_pct,
    daily_loss_total_cap_pct,
    ensure_account_risk_schema,
    risk_control_enabled,
    trading_day_label,
    trading_day_start,
    _parse_open_time_ms,
)
from modules.ctp.vnpy_bridge import (
    ctp_cancel_order,
    ctp_get_tick_price,
    ctp_list_active_orders,
    ctp_list_positions,
    ctp_status,
    execute_order,
)

logger = logging.getLogger(__name__)

CHECK_INTERVAL_SEC = 2
DISCONNECTED_SLEEP_SEC = 5
CLOSED_MARKET_SLEEP_SEC = 30

_flatten_lock = threading.Lock()
_flatten_in_progress = False
_last_flatten_attempt: float = 0.0
FLATTEN_COOLDOWN_SEC = 15


def _closed_in_trading_day(close_time: str, now=None) -> bool:
    oms = _parse_open_time_ms((close_time or "").replace("T", " "))
    if oms is None:
        return False
    return oms >= int(trading_day_start(now).timestamp() * 1000)


def daily_realized_loss_amount(conn, *, now=None) -> float:
    """当日已平仓实现的亏损金额（正数）。"""
    total = 0.0
    try:
        rows = conn.execute(
            "SELECT pnl_net, close_time FROM trade_logs WHERE close_time IS NOT NULL"
        ).fetchall()
    except Exception:
        return 0.0
    for r in rows:
        if isinstance(r, dict):
            ct = r.get("close_time") or ""
            pnl = float(r.get("pnl_net") or 0)
        else:
            ct = r[1] if len(r) > 1 else ""
            pnl = float(r[0] or 0)
        if not _closed_in_trading_day(ct, now):
            continue
        if pnl < 0:
            total += -pnl
    return round(total, 2)


def daily_floating_loss_amount(mode: str) -> float:
    """当前持仓浮亏金额（正数），含隔夜仓跳空。"""
    if not mode:
        return 0.0
    loss = 0.0
    try:
        positions = ctp_list_positions(mode, refresh_if_empty=False, refresh_margin=False)
    except Exception:
        return 0.0
    for p in positions or []:
        lots = int(p.get("lots") or 0)
        if lots <= 0:
            continue
        sym = (p.get("symbol") or "").strip()
        direction = (p.get("direction") or "long").strip().lower()
        entry = float(p.get("avg_price") or p.get("entry_price") or 0)
        if entry <= 0 or not sym:
            continue
        mark = float(p.get("mark_price") or p.get("current_price") or 0)
        if mark <= 0:
            try:
                mark = float(ctp_get_tick_price(mode, sym) or 0)
            except Exception:
                mark = 0.0
        if mark <= 0:
            continue
        m = calc_position_metrics(
            direction, entry, entry, entry, lots, mark, 1.0, sym,
        )
        fp = float(m.get("float_pnl") or 0)
        if fp < 0:
            loss += -fp
    return round(loss, 2)


def daily_loss_amount(
    conn,
    equity: float,
    mode: str,
    *,
    now=None,
) -> tuple[float, float]:
    """返回 (亏损金额, 占权益%)。"""
    if equity <= 0:
        return 0.0, 0.0
    realized = daily_realized_loss_amount(conn, now=now)
    floating = daily_floating_loss_amount(mode)
    total = realized + floating
    pct = round(total / float(equity) * 100, 2)
    return round(total, 2), pct


def daily_loss_used_pct(
    conn,
    equity: float,
    mode: str,
    *,
    now=None,
) -> Optional[float]:
    if equity <= 0:
        return None
    _, pct = daily_loss_amount(conn, equity, mode, now=now)
    return pct


def mark_daily_loss_lock(conn, *, now=None) -> None:
    ensure_account_risk_schema(conn)
    td = trading_day_label(now)
    conn.execute(
        """UPDATE account_risk_state SET trading_day=?, daily_frozen=1,
           cooloff_until_ms=NULL, cooloff_hours=NULL, updated_at=? WHERE id=1""",
        (td, time.strftime("%Y-%m-%d %H:%M:%S")),
    )
    conn.commit()


def _cancel_all_close_orders(mode: str) -> int:
    cancelled = 0
    try:
        active = ctp_list_active_orders(mode)
    except Exception:
        return 0
    for o in active or []:
        offset_s = (o.get("offset") or "").upper()
        if "CLOSE" not in offset_s:
            continue
        oid = str(o.get("order_id") or o.get("vt_order_id") or "")
        if oid and ctp_cancel_order(mode, oid):
            cancelled += 1
    return cancelled


def force_flatten_all_positions(
    conn,
    mode: str,
    *,
    equity: float,
    reason: str = "",
    notify_fn: Callable[[str], None] | None = None,
    get_setting: Callable[[str, str], str] | None = None,
) -> int:
    """无条件市价平掉全部持仓；返回提交平仓笔数。"""
    global _flatten_in_progress, _last_flatten_attempt
    if not ctp_status(mode).get("connected"):
        return 0
    with _flatten_lock:
        if _flatten_in_progress:
            return 0
        if time.time() - _last_flatten_attempt < FLATTEN_COOLDOWN_SEC:
            return 0
        _flatten_in_progress = True
        _last_flatten_attempt = time.time()
    submitted = 0
    try:
        mark_daily_loss_lock(conn)
        cancelled = _cancel_all_close_orders(mode)
        if cancelled:
            logger.info("日亏损强平：已撤平仓挂单 %d 笔", cancelled)
        positions = [
            p for p in (ctp_list_positions(mode) or [])
            if int(p.get("lots") or 0) > 0
        ]
        if not positions:
            return 0
        slip_buf = daily_loss_slippage_buffer_pct(get_setting)
        for p in positions:
            sym = (p.get("symbol") or "").strip()
            direction = (p.get("direction") or "long").strip().lower()
            lots = int(p.get("lots") or 0)
            if not sym or lots <= 0:
                continue
            mark = float(p.get("mark_price") or p.get("current_price") or 0)
            if mark <= 0:
                mark = float(ctp_get_tick_price(mode, sym) or p.get("avg_price") or 0)
            if mark <= 0:
                logger.warning("日亏损强平跳过 %s：无有效价格", sym)
                continue
            offset = "close_long" if direction == "long" else "close_short"
            try:
                execute_order(
                    conn,
                    mode=mode,
                    offset=offset,
                    symbol=sym,
                    direction=direction,
                    lots=lots,
                    price=mark,
                    order_type="market",
                    urgency="risk_flatten",
                    equity=equity,
                    slippage_buffer_pct=slip_buf,
                )
                submitted += 1
                logger.info(
                    "日亏损强平已报单 %s %s %d手 @%s",
                    sym, direction, lots, mark,
                )
            except Exception as exc:
                logger.warning("日亏损强平失败 %s: %s", sym, exc)
        if submitted and notify_fn:
            lim = daily_loss_force_close_pct(get_setting)
            cap = daily_loss_total_cap_pct(get_setting)
            msg = (
                f"日亏损风控：已达权益 {lim:g}% 上限，已强制平仓 {submitted} 笔"
                f"（含滑点预留至 {cap:g}%）。当日禁止开仓。"
            )
            if reason:
                msg = f"{reason} {msg}"
            try:
                notify_fn(msg)
            except Exception as exc:
                logger.debug("daily loss notify: %s", exc)
        if submitted:
            try:
                conn.execute(
                    "UPDATE trade_order_monitors SET status='closed' WHERE status='active'"
                )
                conn.commit()
            except Exception as exc:
                logger.debug("close monitors after flatten: %s", exc)
        return submitted
    finally:
        with _flatten_lock:
            _flatten_in_progress = False


def check_daily_loss_and_flatten(
    conn,
    mode: str,
    *,
    equity: float,
    notify_fn: Callable[[str], None] | None = None,
    get_setting: Callable[[str, str], str] | None = None,
) -> int:
    """达日亏损上限则锁日并强平。返回强平报单笔数。"""
    if not risk_control_enabled():
        return 0
    gs = get_setting or _default_get_setting
    lim = daily_loss_force_close_pct(gs)
    if equity <= 0:
        return 0
    used = daily_loss_used_pct(conn, equity, mode)
    if used is None or used < lim:
        return 0
    amt, pct = daily_loss_amount(conn, equity, mode)
    reason = f"当日亏损 {amt:.0f}元（{pct:.2f}%/权益）"
    return force_flatten_all_positions(
        conn,
        mode,
        equity=equity,
        reason=reason,
        notify_fn=notify_fn,
        get_setting=gs,
    )


def start_daily_loss_guard_worker(
    *,
    db_path: str,
    get_mode_fn: Callable[[], str],
    get_capital_fn: Callable,
    get_setting_fn: Callable[[str, str], str] | None = None,
    init_tables_fn: Callable | None = None,
    notify_fn: Callable[[str], None] | None = None,
    interval: int = CHECK_INTERVAL_SEC,
) -> None:
    from modules.core.db_conn import connect_db

    def _loop() -> None:
        time.sleep(25)
        while True:
            sleep_sec = max(1, interval)
            try:
                mode = get_mode_fn()
                if not ctp_status(mode).get("connected"):
                    time.sleep(DISCONNECTED_SLEEP_SEC)
                    continue
                if not is_trading_session():
                    sleep_sec = max(sleep_sec, CLOSED_MARKET_SLEEP_SEC)
                conn = connect_db(db_path)
                try:
                    if init_tables_fn:
                        init_tables_fn(conn)
                    equity = 0.0
                    try:
                        equity = float(get_capital_fn(conn) or 0)
                    except Exception:
                        equity = 0.0
                    if equity <= 0:
                        try:
                            from modules.ctp.vnpy_bridge import ctp_get_account

                            acc = ctp_get_account(mode) or {}
                            equity = float(acc.get("balance") or 0)
                        except Exception:
                            equity = 0.0
                    if equity > 0:
                        n = check_daily_loss_and_flatten(
                            conn,
                            mode,
                            equity=equity,
                            notify_fn=notify_fn,
                            get_setting=get_setting_fn,
                        )
                        if n:
                            logger.info("日亏损守护: 强制平仓 %d 笔", n)
                finally:
                    conn.close()
            except Exception as exc:
                logger.warning("daily_loss_guard worker: %s", exc)
            time.sleep(sleep_sec)

    threading.Thread(target=_loop, daemon=True, name="daily-loss-guard").start()