live_execution.py

"""
Async execution loop for Binance Futures testnet.
WS for book updates, REST for order placement, delta* lookup from grid.

Testnet book is thin/fake -- the main value is testing control flow
and collecting the decision log (delta* values, order prices).
"""
from __future__ import annotations

import asyncio
import time
import logging
from dataclasses import dataclass, field, asdict
from typing import Any

import numpy as np
from binance import AsyncClient, BinanceSocketManager
from binance.enums import *

logger = logging.getLogger("live_execution")


@dataclass
class FillRecord:
    """Single row in the execution log."""
    timestamp: float = 0.0
    elapsed_sec: float = 0.0
    slice_idx: int = 0
    scheduler: str = ""
    tactic: str = ""
    remaining_qty: float = 0.0
    mid_price: float = 0.0
    best_bid: float = 0.0
    best_ask: float = 0.0
    delta_star: float = 0.0
    order_price: float = 0.0
    order_qty: float = 0.0
    order_side: str = ""
    order_type: str = ""
    fill_price: float = 0.0
    fill_qty: float = 0.0
    is_bps: float = 0.0
    n_levels: int = 0
    action: str = ""  # PLACE_LIMIT, PLACE_MARKET, CANCEL, FILL, SKIP


class LiveExecutor:
    """Async executor for one scheduler+tactic combo on testnet."""

    def __init__(
        self,
        client: AsyncClient,
        cal: dict,
        scheduler_name: str = "AC",
        tactic_name: str = "GLFT",
        check_interval: float = 1.0,
        order_size: float = 0.01,
    ):
        self.client = client
        self.cal = cal
        self.scheduler_name = scheduler_name
        self.tactic_name = tactic_name
        self.check_interval = check_interval
        self.order_size = order_size

        # From calibration
        self.total_qty = cal["total_qty"]
        self.horizon_sec = cal["horizon_sec"]
        self.n_slices = cal["n_slices"]
        self.delta_grid = cal["delta_grid"]
        self.unit_size = cal["unit_size"]
        self.tick_size = cal["tick_size"]
        self.Q = cal["Q"]

        if scheduler_name == "AC":
            self.schedule = cal["ac_schedule"]
        else:
            self.schedule = cal["twap_schedule"]

        # State
        self.remaining = self.total_qty
        self.current_slice = 0
        self.slice_target_rem = self.schedule[0] if len(self.schedule) > 0 else self.total_qty
        self.arrival_price = 0.0
        self.start_time = 0.0

        # Current book state (updated by WebSocket)
        self.best_bid = 0.0
        self.best_ask = 0.0
        self.book_ts = 0.0

        # Pending order tracking
        self.pending_order_id: str | None = None
        self.pending_price = 0.0
        self.pending_qty = 0.0

        # Logs
        self.fills: list[FillRecord] = []
        self.running = False

    def _log(self, **kwargs) -> FillRecord:
        mid = (self.best_bid + self.best_ask) / 2.0 if self.best_bid > 0 else 0
        elapsed = time.time() - self.start_time if self.start_time > 0 else 0

        rec = FillRecord(
            timestamp=time.time(),
            elapsed_sec=elapsed,
            slice_idx=self.current_slice,
            scheduler=self.scheduler_name,
            tactic=self.tactic_name,
            remaining_qty=self.remaining,
            mid_price=mid,
            best_bid=self.best_bid,
            best_ask=self.best_ask,
            **kwargs,
        )
        self.fills.append(rec)
        return rec

    async def _cancel_pending(self):
        if self.pending_order_id is None:
            return
        try:
            await self.client.futures_cancel_order(
                symbol="BTCUSDT",
                orderId=self.pending_order_id,
            )
            self._log(
                action="CANCEL",
                order_price=self.pending_price,
                order_qty=self.pending_qty,
            )
        except Exception as e:
            logger.debug(f"Cancel failed (may already be filled): {e}")
        self.pending_order_id = None

    async def _place_limit_sell(self, price: float, qty: float):
        # Round price to tick, qty to lot
        price = round(price / self.tick_size) * self.tick_size
        qty = round(qty, 3)  # BTCUSDT lot step is 0.001
        qty = max(qty, 0.001)

        try:
            order = await self.client.futures_create_order(
                symbol="BTCUSDT",
                side=SIDE_SELL,
                type=ORDER_TYPE_LIMIT,
                timeInForce=TIME_IN_FORCE_GTC,
                price=f"{price:.2f}",
                quantity=f"{qty:.3f}",
            )
            self.pending_order_id = order["orderId"]
            self.pending_price = price
            self.pending_qty = qty

            self._log(
                action="PLACE_LIMIT",
                order_price=price,
                order_qty=qty,
                order_side="SELL",
                order_type="LIMIT",
                delta_star=self._current_delta(),
            )
        except Exception as e:
            logger.error(f"Limit order failed: {e}")

    async def _place_market_sell(self, qty: float):
        qty = round(qty, 3)
        qty = max(qty, 0.001)

        try:
            order = await self.client.futures_create_order(
                symbol="BTCUSDT",
                side=SIDE_SELL,
                type=ORDER_TYPE_MARKET,
                quantity=f"{qty:.3f}",
            )

            # Parse fills from response
            fill_price = float(order.get("avgPrice", self.best_bid))
            fill_qty = float(order.get("executedQty", qty))

            self.remaining -= fill_qty
            if self.slice_target_rem > 0:
                self.slice_target_rem -= fill_qty
                self.slice_target_rem = max(self.slice_target_rem, 0)

            is_bps = 0.0
            if self.arrival_price > 0:
                is_bps = (self.arrival_price - fill_price) / self.arrival_price * 10000

            self._log(
                action="FILL_MARKET",
                order_price=self.best_bid,
                order_qty=qty,
                order_side="SELL",
                order_type="MARKET",
                fill_price=fill_price,
                fill_qty=fill_qty,
                is_bps=is_bps,
                delta_star=self._current_delta(),
            )

        except Exception as e:
            logger.error(f"Market order failed: {e}")

    async def _check_fills(self):
        if self.pending_order_id is None:
            return

        try:
            order = await self.client.futures_get_order(
                symbol="BTCUSDT",
                orderId=self.pending_order_id,
            )
            status = order.get("status", "")
            executed = float(order.get("executedQty", 0))

            if status == "FILLED" and executed > 0:
                fill_price = float(order.get("avgPrice", self.pending_price))
                self.remaining -= executed
                if self.slice_target_rem > 0:
                    self.slice_target_rem -= executed
                    self.slice_target_rem = max(self.slice_target_rem, 0)

                is_bps = 0.0
                if self.arrival_price > 0:
                    is_bps = (self.arrival_price - fill_price) / self.arrival_price * 10000

                self._log(
                    action="FILL_LIMIT",
                    order_price=self.pending_price,
                    order_qty=self.pending_qty,
                    fill_price=fill_price,
                    fill_qty=executed,
                    is_bps=is_bps,
                )
                self.pending_order_id = None

            elif status in ("CANCELED", "EXPIRED", "REJECTED"):
                self.pending_order_id = None

        except Exception as e:
            logger.debug(f"Order check failed: {e}")

    def _current_delta(self) -> float:
        if self.start_time <= 0:
            return float("inf")

        elapsed = time.time() - self.start_time
        t_idx = int(elapsed)
        t_idx = max(0, min(t_idx, self.delta_grid.shape[1] - 1))

        q_idx = int(round(self.remaining / self.unit_size))
        q_idx = max(1, min(q_idx, self.Q))

        return self.delta_grid[q_idx, t_idx]

    def _advance_slice(self):
        if self.start_time <= 0:
            return

        elapsed = time.time() - self.start_time
        slice_dur = self.horizon_sec / self.n_slices
        expected = min(int(elapsed / slice_dur), self.n_slices - 1)

        while self.current_slice < expected:
            carry = self.slice_target_rem
            self.current_slice += 1
            if self.current_slice < self.n_slices:
                self.slice_target_rem = self.schedule[self.current_slice] + carry
            else:
                self.slice_target_rem = self.remaining

    async def run(self):
        """Main loop, runs for horizon_sec."""
        self.running = True
        self.start_time = time.time()

        # Start WebSocket for book ticker
        bm = BinanceSocketManager(self.client)
        ts = bm.symbol_ticker_futures_socket("BTCUSDT")

        async with ts as stream:
            # Run the control loop concurrently with WS updates
            ws_task = asyncio.create_task(self._ws_reader(stream))
            control_task = asyncio.create_task(self._control_loop())

            try:
                await asyncio.gather(ws_task, control_task)
            except asyncio.CancelledError:
                pass
            finally:
                ws_task.cancel()
                self.running = False

    async def _ws_reader(self, stream):
        while self.running:
            try:
                msg = await asyncio.wait_for(stream.recv(), timeout=5.0)
                # futures ticker has 'data' wrapper or direct fields
                data = msg.get("data", msg) if isinstance(msg, dict) else msg
                if isinstance(data, dict):
                    bid = data.get("b", data.get("bidPrice"))
                    ask = data.get("a", data.get("askPrice"))
                    if bid and ask:
                        self.best_bid = float(bid)
                        self.best_ask = float(ask)
                        self.book_ts = time.time()
            except asyncio.TimeoutError:
                continue
            except Exception as e:
                logger.error(f"WS error: {e}")
                break

            
    async def _control_loop(self):
        """1Hz control loop: advance slice, check fills, place/cancel orders."""
        while self.running:
            await asyncio.sleep(self.check_interval)

            elapsed = time.time() - self.start_time

            # Wait for valid book data
            if self.best_bid <= 0 or self.best_ask <= 0:
                continue

            mid = (self.best_bid + self.best_ask) / 2.0

            # Set arrival price on first valid tick
            if self.arrival_price <= 0:
                self.arrival_price = mid
                logger.info(f"Arrival price: {mid:.2f}")

            # Check if done
            if self.remaining <= 0.001:
                logger.info("Fully liquidated")
                break

            # Past horizon: terminal dump
            if elapsed >= self.horizon_sec:
                logger.info(f"Terminal dump: {self.remaining:.3f} BTC")
                await self._cancel_pending()
                await self._place_market_sell(self.remaining)
                break

            # Advance slice
            self._advance_slice()

            # Check fills on pending order
            await self._check_fills()

            # Skip if nothing to do this slice
            if self.slice_target_rem <= 0.001 or self.remaining <= 0.001:
                self._log(action="SKIP")
                continue

            # ── Tactic decision ────────────────────────────────────
            if self.tactic_name == "MARKET":
                # Market order tactic: sell slice target immediately
                qty = min(self.slice_target_rem, self.remaining)
                await self._place_market_sell(qty)
                self.slice_target_rem = 0

            elif self.tactic_name == "PEG":
                # Peg best ask
                qty = min(self.order_size, self.slice_target_rem, self.remaining)
                if self.pending_order_id is not None:
                    if abs(self.pending_price - self.best_ask) > self.tick_size / 2:
                        await self._cancel_pending()
                    else:
                        continue  # order is at best ask, wait

                await self._place_limit_sell(self.best_ask, qty)

            elif self.tactic_name == "GLFT":
                # GLFT: look up delta*, decide limit vs market
                delta_star = self._current_delta()
                optimal_ask = mid + delta_star
                optimal_ask = round(optimal_ask / self.tick_size) * self.tick_size

                qty = min(self.order_size, self.slice_target_rem, self.remaining)

                if delta_star <= 0 or optimal_ask <= self.best_bid:
                    # Aggressive: cross the spread
                    await self._cancel_pending()
                    await self._place_market_sell(qty)
                else:
                    # Passive: post at optimal ask
                    if self.pending_order_id is not None:
                        if abs(self.pending_price - optimal_ask) > self.tick_size:
                            await self._cancel_pending()
                        else:
                            continue

                    await self._place_limit_sell(optimal_ask, qty)

        # End of execution
        logger.info(f"Execution complete. Remaining: {self.remaining:.4f} BTC")
        logger.info(f"Total fills logged: {len(self.fills)}")