qihuo/strategy/strategy_roll_lib.py

"""顺势加仓（滚仓）：纯计算，期货版（手数整数、乘数计入盈亏）。"""
from __future__ import annotations

import math
from typing import Any, Optional, Tuple

from strategy.fib_lib import calc_fib_plan

ROLL_MAX_LEGS_LONG = 3
ROLL_MAX_LEGS_SHORT = 3
ROLL_STOP_OFFSET_PCT_DEFAULT = 1.0
FIB_MODES = frozenset({"fib_618", "fib_786"})


def fib_ratio_from_mode(mode: str) -> Optional[float]:
    m = (mode or "").strip().lower()
    if m in ("fib_618", "618", "0.618"):
        return 0.618
    if m in ("fib_786", "786", "0.786"):
        return 0.786
    return None


def fib_limit_entry(direction: str, upper: float, lower: float, mode: str) -> Tuple[Optional[float], Optional[str]]:
    ratio = fib_ratio_from_mode(mode)
    if ratio is None:
        return None, "斐波档位无效"
    h, l = float(upper), float(lower)
    if h <= l:
        return None, "上沿须大于下沿"
    direction = (direction or "long").strip().lower()
    plan = calc_fib_plan(direction, h, l, ratio)
    if not plan:
        return None, "无法计算斐波限价"
    entry, _sl, _tp = plan
    return float(entry), None


def max_roll_legs(direction: str) -> int:
    return ROLL_MAX_LEGS_LONG if (direction or "long").strip().lower() == "long" else ROLL_MAX_LEGS_SHORT


def lots_precise(raw: float) -> int:
    if raw is None or raw < 1:
        return 0
    return max(1, int(math.floor(float(raw))))


def unified_stop_from_avg(direction: str, avg: float, offset_pct: float) -> float:
    avg_f = float(avg)
    pct = float(offset_pct) / 100.0
    direction = (direction or "long").strip().lower()
    if direction == "short":
        return avg_f * (1.0 + pct)
    return avg_f * (1.0 - pct)


def avg_entry_after_add(qty_existing: float, entry_existing: float, add_qty: float, add_price: float) -> float:
    q1, e1, q2, e2 = float(qty_existing), float(entry_existing), float(add_qty), float(add_price)
    total = q1 + q2
    return (q1 * e1 + q2 * e2) / total if total > 0 else 0.0


def solve_add_lots_for_total_risk(
    direction: str,
    qty_existing: float,
    entry_existing: float,
    add_price: float,
    new_stop: float,
    risk_budget: float,
    mult: int,
) -> Tuple[Optional[int], Optional[str]]:
    q1, e1, e2, sl, b = float(qty_existing), float(entry_existing), float(add_price), float(new_stop), float(risk_budget)
    m = float(mult)
    direction = (direction or "long").strip().lower()
    if direction == "short":
        denom = (sl - e2) * m
        numer = b - q1 * (sl - e1) * m
    else:
        denom = (e2 - sl) * m
        numer = b - q1 * (e1 - sl) * m
    if denom <= 0:
        return None, "止损与加仓价关系无效"
    q2 = numer / denom
    lots = lots_precise(q2)
    if lots < 1:
        return None, "按总风险%无需再加仓或无法再加"
    return lots, None


def preview_roll(
    *,
    direction: str,
    symbol: str,
    qty_existing: float,
    entry_existing: float,
    initial_take_profit: float,
    add_mode: str,
    new_stop_loss: float,
    risk_percent: float,
    capital_base: float,
    mult: int,
    add_price: Optional[float] = None,
    fib_upper: Optional[float] = None,
    fib_lower: Optional[float] = None,
    legs_done: int = 0,
) -> Tuple[Optional[dict[str, Any]], Optional[str]]:
    direction = (direction or "long").strip().lower()
    if legs_done >= max_roll_legs(direction):
        return None, f"滚仓已达 {max_roll_legs(direction)} 次上限"
    mode = (add_mode or "market").strip().lower()
    if mode == "market":
        if not add_price or add_price <= 0:
            return None, "需要有效参考价"
        entry_add = float(add_price)
        mode_label = "市价"
    elif mode in FIB_MODES:
        if fib_upper is None or fib_lower is None:
            return None, "斐波须填上沿/下沿"
        entry_add, err = fib_limit_entry(direction, float(fib_upper), float(fib_lower), mode)
        if err:
            return None, err
        mode_label = "斐波0.618" if "618" in mode else "斐波0.786"
    else:
        return None, "加仓方式无效"
    sl = float(new_stop_loss)
    tp = float(initial_take_profit)
    if sl <= 0 or tp <= 0:
        return None, "止损/止盈无效"
    risk_budget = float(capital_base) * float(risk_percent) / 100.0
    q2, err = solve_add_lots_for_total_risk(
        direction, qty_existing, entry_existing, entry_add, sl, risk_budget, mult
    )
    if err:
        return None, err
    new_qty = qty_existing + q2
    new_avg = avg_entry_after_add(qty_existing, entry_existing, q2, entry_add)
    m = float(mult)
    if direction == "long":
        loss_at_sl = (new_avg - sl) * new_qty * m
        reward_at_tp = (tp - new_avg) * new_qty * m
    else:
        loss_at_sl = (sl - new_avg) * new_qty * m
        reward_at_tp = (new_avg - tp) * new_qty * m
    return {
        "symbol": symbol,
        "direction": direction,
        "add_mode_label": mode_label,
        "add_price": round(entry_add, 4),
        "new_stop_loss": round(sl, 4),
        "initial_take_profit": tp,
        "risk_percent": float(risk_percent),
        "add_lots": q2,
        "qty_after": int(new_qty),
        "avg_entry_after": round(new_avg, 4),
        "loss_at_sl": round(loss_at_sl, 2),
        "reward_at_tp": round(reward_at_tp, 2),
        "legs_done": legs_done,
    }, None