strategy/builder.py

from __future__ import annotations

from market_data.schemas import OptionChain, OptionContract

from .payoff import estimate_breakevens
from .schemas import OptionLeg, OptionStrategy


def usable_contracts(contracts: list[OptionContract]) -> list[OptionContract]:
    return [
        contract
        for contract in contracts
        if contract.mid is not None
        and contract.mid > 0
        and not {"missing_or_zero_bid_ask", "zero_open_interest"}.intersection(contract.liquidity_warnings)
    ]


def nearest_contract(contracts: list[OptionContract], target_strike: float) -> OptionContract | None:
    valid = usable_contracts(contracts)
    if not valid:
        return None
    return min(valid, key=lambda contract: abs(contract.strike - target_strike))


def contract_to_leg(contract: OptionContract, action: str, quantity: int = 1) -> OptionLeg:
    return OptionLeg(
        action=action,
        option_type=contract.option_type,
        strike=contract.strike,
        expiration=contract.expiration,
        quantity=quantity,
        premium=contract.mid or contract.last_price or 0.0,
        implied_volatility=contract.implied_volatility,
        liquidity_warnings=contract.liquidity_warnings,
    )


def make_strategy(
    name: str,
    volatility_view: str,
    directional_view: str,
    legs: list[OptionLeg],
    rationale: str,
    risks: list[str],
    score: float,
) -> OptionStrategy:
    net_cash_flow = sum(leg.cash_flow() for leg in legs)
    net_debit_or_credit = -net_cash_flow
    breakevens = estimate_breakevens(legs)

    max_profit: float | str | None = None
    max_loss: float | str | None = None
    if name in {"long_straddle", "long_strangle"}:
        max_loss = round(max(net_debit_or_credit, 0.0), 2)
        max_profit = "unlimited"
    elif name == "short_straddle":
        max_profit = round(abs(min(net_debit_or_credit, 0.0)), 2)
        max_loss = "unlimited"
    elif name == "iron_condor":
        call_strikes = sorted(leg.strike for leg in legs if leg.option_type == "call")
        put_strikes = sorted(leg.strike for leg in legs if leg.option_type == "put")
        width = max(call_strikes[-1] - call_strikes[0], put_strikes[-1] - put_strikes[0])
        credit = abs(min(net_debit_or_credit, 0.0))
        max_profit = round(credit, 2)
        max_loss = round(width * 100 - credit, 2)
    elif name == "calendar_spread":
        max_loss = round(max(net_debit_or_credit, 0.0), 2)
        max_profit = "path_dependent"

    return OptionStrategy(
        name=name,
        volatility_view=volatility_view,
        directional_view=directional_view,
        legs=legs,
        rationale=rationale,
        risks=risks,
        max_profit=max_profit,
        max_loss=max_loss,
        breakevens=breakevens,
        net_debit_or_credit=round(net_debit_or_credit, 2),
        score=score,
    )


def generate_volatility_strategies(
    near_chain: OptionChain,
    volatility_view: str = "neutral",
    directional_view: str = "neutral",
    far_chain: OptionChain | None = None,
) -> list[OptionStrategy]:
    if near_chain.underlying_price is None:
        return []

    spot = near_chain.underlying_price
    atm_call = nearest_contract(near_chain.calls, spot)
    atm_put = nearest_contract(near_chain.puts, spot)
    otm_call = nearest_contract(near_chain.calls, spot * 1.05)
    otm_put = nearest_contract(near_chain.puts, spot * 0.95)
    strategies: list[OptionStrategy] = []

    if atm_call and atm_put:
        if volatility_view in {"long_vol", "neutral", "vol_expansion"}:
            strategies.append(
                make_strategy(
                    name="long_straddle",
                    volatility_view="long_vol",
                    directional_view="neutral",
                    legs=[contract_to_leg(atm_call, "buy"), contract_to_leg(atm_put, "buy")],
                    rationale="Benefits from a large realized move or IV expansion; risk is premium paid.",
                    risks=["theta_decay", "iv_crush", "requires_large_move"],
                    score=0.75,
                )
            )
        if volatility_view in {"short_vol", "neutral", "vol_compression"}:
            strategies.append(
                make_strategy(
                    name="short_straddle",
                    volatility_view="short_vol",
                    directional_view="neutral",
                    legs=[contract_to_leg(atm_call, "sell"), contract_to_leg(atm_put, "sell")],
                    rationale="Benefits from realized volatility staying below implied volatility.",
                    risks=["unlimited_tail_risk", "gap_risk", "margin_requirement"],
                    score=0.45,
                )
            )

    if otm_call and otm_put and volatility_view in {"long_vol", "neutral", "vol_expansion"}:
        strategies.append(
            make_strategy(
                name="long_strangle",
                volatility_view="long_vol",
                directional_view="neutral",
                legs=[contract_to_leg(otm_call, "buy"), contract_to_leg(otm_put, "buy")],
                rationale="Lower-cost long volatility expression than a straddle, but needs a larger move.",
                risks=["theta_decay", "wide_breakevens", "iv_crush"],
                score=0.65,
            )
        )

    if far_chain and atm_call and volatility_view in {"long_vol", "neutral", "term_structure"}:
        far_call = nearest_contract(far_chain.calls, atm_call.strike)
        if far_call:
            strategies.append(
                make_strategy(
                    name="calendar_spread",
                    volatility_view="term_structure",
                    directional_view="neutral",
                    legs=[contract_to_leg(atm_call, "sell"), contract_to_leg(far_call, "buy")],
                    rationale="Expresses a term-structure view and benefits if longer-dated IV holds up.",
                    risks=["path_dependency", "front_expiry_gamma", "term_structure_shift"],
                    score=0.60,
                )
            )

    if otm_call and otm_put and volatility_view in {"short_vol", "neutral", "vol_compression"}:
        long_call = nearest_contract(near_chain.calls, otm_call.strike * 1.03)
        long_put = nearest_contract(near_chain.puts, otm_put.strike * 0.97)
        if long_call and long_put:
            strategies.append(
                make_strategy(
                    name="iron_condor",
                    volatility_view="short_vol",
                    directional_view="neutral",
                    legs=[
                        contract_to_leg(otm_put, "sell"),
                        contract_to_leg(long_put, "buy"),
                        contract_to_leg(otm_call, "sell"),
                        contract_to_leg(long_call, "buy"),
                    ],
                    rationale="Defined-risk short volatility strategy for range-bound markets.",
                    risks=["short_gamma", "tail_loss_to_width", "assignment_risk"],
                    score=0.70,
                )
            )

    return sorted(strategies, key=lambda strategy: strategy.score, reverse=True)