qihuo/strategy/strategy_roll_monitor_lib.py

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

"""顺势滚仓程序监控：突破 pending 腿触价成交、外部平仓同步。"""
from __future__ import annotations

import logging
from datetime import datetime
from typing import Any, Callable, Optional
from zoneinfo import ZoneInfo

from contract_specs import get_contract_spec
from strategy.strategy_roll_lib import (
    ADD_MODE_BREAKOUT,
    FIB_MODES,
    LEG_STATUS_CANCELLED,
    LEG_STATUS_FILLED,
    LEG_STATUS_INVALIDATED,
    LEG_STATUS_PENDING,
    detect_mark_cross,
    preview_roll,
)

logger = logging.getLogger(__name__)
TZ = ZoneInfo("Asia/Shanghai")


def _now() -> str:
    return datetime.now(TZ).strftime("%Y-%m-%d %H:%M:%S")


def roll_sync_after_external_close(conn, *, monitor_id: int) -> None:
    """手动平仓或监控结案后关闭滚仓组并清除 pending 腿。"""
    grp = conn.execute(
        "SELECT id FROM roll_groups WHERE order_monitor_id=? AND status='active'",
        (int(monitor_id),),
    ).fetchone()
    if not grp:
        return
    gid = int(grp["id"])
    conn.execute(
        "UPDATE roll_legs SET status=? WHERE roll_group_id=? AND status=?",
        (LEG_STATUS_CANCELLED, gid, LEG_STATUS_PENDING),
    )
    conn.execute(
        "UPDATE roll_groups SET status='closed', updated_at=? WHERE id=?",
        (_now(), gid),
    )


def cancel_roll_leg(conn, leg_id: int) -> tuple[bool, str]:
    row = conn.execute(
        "SELECT * FROM roll_legs WHERE id=? AND status=?",
        (int(leg_id), LEG_STATUS_PENDING),
    ).fetchone()
    if not row:
        return False, "仅可删除监控中的腿"
    conn.execute(
        "UPDATE roll_legs SET status=? WHERE id=?",
        (LEG_STATUS_CANCELLED, int(leg_id)),
    )
    return True, "已删除"


def check_roll_monitors(
    conn,
    *,
    get_mark_price_fn: Callable[[str], Optional[float]],
    fill_roll_leg_fn: Callable[[dict, dict, dict, dict], tuple[bool, str]],
    is_trading_session_fn: Callable[[], bool],
    get_risk_budget_fn: Callable[[], float],
) -> None:
    """扫描 pending 滚仓腿，标记价穿越则重算手数并市价成交。"""
    if not is_trading_session_fn():
        return
    rows = conn.execute(
        """SELECT l.*, g.order_monitor_id, g.symbol, g.direction, g.initial_take_profit,
                  g.risk_percent, g.leg_count AS group_leg_count,
                  m.lots AS mon_lots, m.entry_price AS mon_entry, m.take_profit AS mon_tp,
                  m.status AS mon_status
           FROM roll_legs l
           JOIN roll_groups g ON g.id = l.roll_group_id
           JOIN trade_order_monitors m ON m.id = g.order_monitor_id
           WHERE l.status=? AND g.status='active' AND m.status='active'""",
        (LEG_STATUS_PENDING,),
    ).fetchall()
    for raw in rows:
        leg = dict(raw)
        if (leg.get("mon_status") or "").strip().lower() != "active":
            _invalidate_leg(conn, leg, "监控已结束")
            continue
        sym = (leg.get("symbol") or "").strip()
        mark = get_mark_price_fn(sym)
        if not mark or mark <= 0:
            continue
        prev_mark = float(leg.get("last_mark_price") or mark)
        mode = (leg.get("add_mode") or "").strip().lower()
        trigger = float(leg.get("limit_price") or leg.get("breakthrough_price") or 0)
        direction = (leg.get("direction") or "long").strip().lower()
        if mode in FIB_MODES or mode == ADD_MODE_BREAKOUT:
            if not detect_mark_cross(direction, mode, prev_mark, mark, trigger):
                conn.execute(
                    "UPDATE roll_legs SET last_mark_price=? WHERE id=?",
                    (float(mark), int(leg["id"])),
                )
                continue
        mon = {
            "id": leg["order_monitor_id"],
            "symbol": sym,
            "direction": direction,
            "lots": leg["mon_lots"],
            "entry_price": leg["mon_entry"],
            "take_profit": leg["mon_tp"] or leg["initial_take_profit"],
        }
        grp = {
            "id": leg["roll_group_id"],
            "order_monitor_id": leg["order_monitor_id"],
            "leg_count": leg.get("group_leg_count") or 0,
            "risk_percent": leg.get("risk_percent"),
        }
        preview, err = preview_roll(
            direction=direction,
            symbol=sym,
            qty_existing=float(leg["mon_lots"] or 0),
            entry_existing=float(leg["mon_entry"] or 0),
            initial_take_profit=float(leg["mon_tp"] or leg["initial_take_profit"] or 0),
            add_mode=mode,
            new_stop_loss=float(leg["new_stop_loss"] or 0),
            risk_budget=float(leg.get("risk_percent") or 0) or get_risk_budget_fn(),
            mult=int(get_contract_spec(sym).get("mult") or 1),
            mark_price=mark,
            limit_price=trigger if mode in FIB_MODES else None,
            breakthrough_price=trigger if mode == ADD_MODE_BREAKOUT else None,
            legs_done=int(leg.get("group_leg_count") or 0),
            at_trigger=True,
        )
        if err or not preview:
            _invalidate_leg(conn, leg, err or "触发时无法加仓")
            continue
        ok, msg = fill_roll_leg_fn(mon, grp, leg, preview)
        if not ok:
            logger.warning("roll leg fill failed #%s: %s", leg.get("id"), msg)


def _invalidate_leg(conn, leg: dict, reason: str) -> None:
    conn.execute(
        "UPDATE roll_legs SET status=?, invalidated_reason=? WHERE id=?",
        (LEG_STATUS_INVALIDATED, (reason or "")[:200], int(leg["id"])),
    )