v3_execution_engine.py

#!/usr/bin/env python3
"""Strict Bitget next-quote paper execution adapter for V3 shadow events."""

from __future__ import annotations

import csv
import json
from pathlib import Path
from typing import Any

import pandas as pd

from common import BASE_DIR, bitget_contract_ticker
from instrument_registry import validate_instrument


DEFAULT_SHARED_ROOT = BASE_DIR
REGULAR_QUOTE_MAX_AGE_SECONDS = 60
EXTENDED_QUOTE_MAX_AGE_SECONDS = 300
UNDERLYING_1M_DIR = BASE_DIR / "data" / "v3_point_in_time" / "us_underlying" / "1m"


def utc(value: Any) -> pd.Timestamp:
    stamp = pd.Timestamp(value)
    return stamp.tz_localize("UTC") if stamp.tzinfo is None else stamp.tz_convert("UTC")


def _read_json(path: Path) -> dict[str, Any]:
    return json.loads(path.read_text(encoding="utf-8-sig"))


def _corporate_action_covered(ticker: str, quote_time: pd.Timestamp, shared_root: Path) -> bool:
    path = shared_root / "data" / "point_in_time_features" / "corporate_actions" / "observations_v1.csv"
    if not path.exists():
        return False
    with path.open(encoding="utf-8-sig", newline="") as handle:
        rows = csv.DictReader(handle)
        for row in rows:
            if str(row.get("ticker") or "").upper() != ticker:
                continue
            if "split" not in str(row.get("feature_name") or "").lower():
                continue
            if str(row.get("status") or "").lower() != "captured":
                continue
            available = pd.to_datetime(row.get("available_time"), utc=True, errors="coerce")
            if not pd.isna(available) and available <= quote_time:
                return True
    return False


def _latest_underlying_reference(ticker: str, now: pd.Timestamp) -> dict[str, Any] | None:
    path = UNDERLYING_1M_DIR / f"{ticker}.csv"
    if not path.exists():
        return None
    frame = pd.read_csv(path, encoding="utf-8-sig")
    required = {"bar_close_time", "available_time", "close"}
    if not required.issubset(frame.columns):
        return None
    for column in ("bar_close_time", "available_time"):
        frame[column] = pd.to_datetime(frame[column], utc=True, errors="coerce", format="mixed")
    frame["close"] = pd.to_numeric(frame["close"], errors="coerce")
    frame = frame[
        (frame["bar_close_time"] <= now)
        & (frame["available_time"] <= now)
        & (frame["close"] > 0)
    ].dropna(subset=["bar_close_time", "available_time", "close"]).sort_values("bar_close_time")
    if frame.empty:
        return None
    latest = frame.iloc[-1]
    age = float((now - latest["bar_close_time"]).total_seconds())
    phase = "regular" if 13 <= now.hour < 21 else "extended"
    maximum_age = REGULAR_QUOTE_MAX_AGE_SECONDS * 3 if phase == "regular" else EXTENDED_QUOTE_MAX_AGE_SECONDS
    if age < 0 or age > maximum_age:
        return None
    closes = frame["close"].tail(1500)
    no_scale_jump = len(closes) >= 20 and not closes.pct_change().abs().gt(0.35).any()
    return {
        "price": float(latest["close"]),
        "timestamp": latest["bar_close_time"].isoformat(),
        "source": "V3 latest completed underlying 1m bar",
        "age_seconds": age,
        "no_scale_jump": bool(no_scale_jump),
        "path": str(path),
    }


def load_validated_execution_quote(
    *,
    ticker: str,
    direction: str,
    decision_time: Any,
    underlying_price: float,
    underlying_time: Any,
    shared_root: Path = DEFAULT_SHARED_ROOT,
    observed_at: Any | None = None,
) -> dict[str, Any]:
    ticker = str(ticker).upper().strip()
    direction = str(direction).upper()
    if direction not in {"LONG", "SHORT"}:
        raise ValueError("v3_execution_direction_required")
    watchlist_path = shared_root / "data" / "bitget_us_equity_contract_watchlist.csv"
    metadata_path = shared_root / "data" / "bitget_us_equity_contract_watchlist_metadata.json"
    specs_path = shared_root / "data" / "bitget_contract_specs_usdt-futures.json"
    if not all(path.exists() for path in (watchlist_path, metadata_path, specs_path)):
        raise RuntimeError("bitget_execution_snapshot_or_specs_missing")
    with watchlist_path.open(encoding="utf-8-sig", newline="") as handle:
        row = next((item for item in csv.DictReader(handle) if str(item.get("ticker")).upper() == ticker), None)
    if row is None:
        raise RuntimeError(f"bitget_execution_contract_missing:{ticker}")
    specs_payload = _read_json(specs_path)
    symbol = str(row.get("symbol") or "").upper()
    spec = (specs_payload.get("contracts") or {}).get(symbol) or {}
    decision = utc(decision_time)
    now = utc(observed_at) if observed_at is not None else pd.Timestamp.now(tz="UTC")
    live_underlying = _latest_underlying_reference(ticker, now)
    if live_underlying is not None:
        underlying_price = float(live_underlying["price"])
        underlying_time = live_underlying["timestamp"]
    phase = "regular" if 13 <= now.hour < 21 else "extended"
    maximum_age = REGULAR_QUOTE_MAX_AGE_SECONDS if phase == "regular" else EXTENDED_QUOTE_MAX_AGE_SECONDS
    try:
        live = bitget_contract_ticker(ticker)
    except Exception as exc:
        raise RuntimeError(
            f"bitget_live_quote_refresh_failed:{type(exc).__name__}:{str(exc)[:160]}"
        ) from exc
    if not live or str(live.get("symbol") or "").upper() != symbol:
        raise RuntimeError(f"bitget_live_contract_missing:{ticker}")
    if live.get("time_provenance_verified") is not True:
        raise RuntimeError("bitget_quote_time_provenance_unverified")
    provider_quote_time = utc(live.get("provider_timestamp"))
    request_started_at = utc(live.get("request_started_at"))
    received_at = utc(live.get("received_at"))
    if request_started_at < decision:
        raise RuntimeError("bitget_quote_request_started_before_decision")
    if provider_quote_time <= decision:
        raise RuntimeError("bitget_provider_quote_not_after_decision")
    if received_at < provider_quote_time or received_at < request_started_at:
        raise RuntimeError("bitget_quote_clock_order_invalid")
    quote_age = float((received_at - provider_quote_time).total_seconds())
    if quote_age < 0 or quote_age > maximum_age:
        raise RuntimeError(f"bitget_quote_stale:{quote_age:.3f}")
    bid = float(live.get("bid_price") or 0.0)
    ask = float(live.get("ask_price") or 0.0)
    quote_source = str(live.get("source") or "Bitget public mix ticker")
    if min(bid, ask) <= 0 or ask < bid:
        raise RuntimeError("bitget_bid_ask_invalid")
    execution_price = ask if direction == "LONG" else bid
    registered_action = _corporate_action_covered(ticker, provider_quote_time, shared_root)
    locally_normalized = bool(live_underlying and live_underlying.get("no_scale_jump"))
    mapping = {
        "execution_symbol": symbol,
        "contract_specification_verified": bool(spec) and spec.get("symbolStatus") == "normal",
        "contract_multiplier": spec.get("sizeMultiplier"),
        "contract_quantity_unit": spec.get("baseCoin"),
        "contract_quantity_step": spec.get("minTradeNum"),
        "contract_taker_fee_rate": spec.get("takerFeeRate"),
        "contract_maker_fee_rate": spec.get("makerFeeRate"),
        "contract_spec_source": str(specs_path),
        "contract_spec_product_type": specs_payload.get("product_type"),
        "contract_symbol_status": spec.get("symbolStatus"),
        "contract_symbol_type": spec.get("symbolType"),
        "contract_is_rwa": spec.get("isRwa") == "YES",
        "corporate_action_normalized": registered_action or locally_normalized,
        "corporate_action_normalization_method": (
            "prospective_split_observation_present" if registered_action
            else "v3_completed_1m_no_scale_jump" if locally_normalized
            else "unverified"
        ),
    }
    signal_frame = pd.DataFrame([{
        "close": float(underlying_price),
        "datetime": utc(underlying_time).isoformat(),
        "data_source": "V3 completed underlying intraday bar",
    }])
    quote = {
        "symbol": symbol,
        "price": execution_price,
        "bid_price": bid,
        "ask_price": ask,
        "timestamp": provider_quote_time.isoformat(),
        "equity_reference_price": float(underlying_price),
        "equity_reference_timestamp": utc(underlying_time).isoformat(),
        "equity_reference_source": (
            live_underlying["source"] if live_underlying is not None
            else "V3 completed underlying intraday bar"
        ),
        "source": quote_source,
        "contract_taker_fee_rate": spec.get("takerFeeRate"),
        "contract_maker_fee_rate": spec.get("makerFeeRate"),
        **mapping,
    }
    validation = validate_instrument(
        ticker,
        signal_frame=signal_frame,
        bitget_quote=quote,
        mapping=mapping,
        require_bitget=True,
    )
    return {
        "quote_time": received_at.isoformat(),
        "quote_price": execution_price,
        "quote_source": quote_source,
        "directional_price_side": "ask" if direction == "LONG" else "bid",
        "quote_age_seconds": quote_age,
        "quote_max_age_seconds": maximum_age,
        "instrument_validation": validation,
        "underlying_reference": live_underlying,
        "execution_provenance": {
            "provider_quote_time": provider_quote_time.isoformat(),
            "request_started_at": request_started_at.isoformat(),
            "received_at": received_at.isoformat(),
            "execution_available_time": received_at.isoformat(),
            "provider_time_field": "provider_timestamp",
            "time_provenance_verified": True,
        },
    }


def build_risk_plan(entry_price: float, direction: str, config_path: Path | None = None) -> dict[str, Any]:
    path = config_path or (BASE_DIR / "paper_config.json")
    config = _read_json(path)
    if config.get("paper_trading_only") is not True:
        raise RuntimeError("v3_risk_plan_requires_paper_only_config")
    stop_pct = float(config["stop_loss_pct"])
    target_pct = float(config["take_profit_pct"])
    if stop_pct != -0.04 or target_pct != 0.08:
        raise RuntimeError("v3_risk_parameters_changed_from_frozen_policy")
    max_holding_days = int(config["max_holding_days"])
    stale_unprofitable_days = int(config.get("stale_unprofitable_days", 4))
    if stale_unprofitable_days < 1 or stale_unprofitable_days >= max_holding_days:
        raise RuntimeError("v3_stale_exit_days_must_precede_hard_time_stop")
    entry = float(entry_price)
    side = str(direction).upper()
    if side == "LONG":
        stop, target = entry * (1.0 + stop_pct), entry * (1.0 + target_pct)
    elif side == "SHORT":
        stop, target = entry * (1.0 - stop_pct), entry * (1.0 - target_pct)
    else:
        raise ValueError("v3_risk_plan_direction_invalid")
    return {
        "stop_loss": round(stop, 8),
        "take_profit": round(target, 8),
        "stop_loss_pct": stop_pct,
        "take_profit_pct": target_pct,
        "trailing_stop_pct": float(config["trailing_stop_pct"]),
        "stale_unprofitable_days": stale_unprofitable_days,
        "max_holding_days": max_holding_days,
        "holding_period_unit": str(config["holding_period_unit"]),
        "invalidation": (
            f"触发止损，或持有满{stale_unprofitable_days}天仍不盈利，"
            f"或达到{max_holding_days}个日历天时间止损，或原始数据门失效"
        ),
        "paper_trading_only": True,
    }