Algo Expert

"""
ATR Breakout - dual-mode strategy with REAL-TIME LIMIT execution.

Computes a volatility band: prior close +/- ATR_MULTIPLIER * ATR(N).
Pre-places a LIMIT BUY at upper band and LIMIT SELL at lower band.
Modifies the LIMIT prices on each tick as ATR/close drifts.

If the LIMIT isn't filled within LIMIT_TIMEOUT_SEC after a band touch,
falls back to MARKET to guarantee execution.

Backtest: applies COSTS.slippage to model the LIMIT/MARKET fallback ratio.
"""
import argparse, logging, os, signal, sys, threading, time
from datetime import datetime, timedelta
from pathlib import Path

import pandas as pd
from dotenv import find_dotenv, load_dotenv

_HERE = Path(__file__).resolve().parent
for parent in [_HERE, *_HERE.parents]:
    candidate = parent / ".claude" / "skills" / "algo-expert" / "rules" / "assets" / "core"
    if candidate.exists():
        sys.path.insert(0, str(candidate.parent)); break

from openalgo import api  # noqa: E402
from core.cost_model import lookup as cost_lookup, format_cost_report, SlippageTracker  # noqa: E402
from core.indicator_adapter import get_indicators  # noqa: E402
from core.data_router import fetch_backtest_data, warmup_live_data  # noqa: E402
from core.risk_manager import RiskManager, RiskConfig, Position  # noqa: E402
from core.sizing import fixed_fractional_size, compute_live_qty  # noqa: E402
from core.preflight import run_preflight, find_existing_open_position  # noqa: E402
from core.state import StrategyState  # noqa: E402

# === Config ===
SYMBOL          = "RELIANCE"
EXCHANGE        = os.getenv("OPENALGO_STRATEGY_EXCHANGE", os.getenv("EXCHANGE", "NSE"))
INTERVAL        = "5m"
PRODUCT         = "MIS"
LOT_SIZE        = 1
STRATEGY_NAME   = os.getenv("STRATEGY_NAME", "atr_breakout")
DATA_SOURCE     = os.getenv("DATA_SOURCE", "api")
RISK_PER_TRADE  = 0.005
MAX_SIZE_PCT    = 0.50
ATR_PERIOD      = 14
ATR_MULTIPLIER  = 1.5
INDICATOR_LIB   = "openalgo"
EXECUTION_TYPE  = "limit"

LIMIT_OFFSET_PCT      = 0.0005    # peg LIMIT this far from band
LIMIT_TIMEOUT_SEC     = 3         # fall back to MARKET if not filled
LIMIT_MODIFY_THROTTLE = 1.5       # min seconds between modifies
SQUARE_OFF_TIME_HOUR  = 15
SQUARE_OFF_TIME_MIN   = 15

RISK = RiskConfig(sl_pct=0.012, tp_pct=0.025, trail_pct=0.01, time_exit_min=240)
INIT_CASH       = 1_000_000
LOOKBACK_DAYS   = 365 * 2

logging.basicConfig(level=os.getenv("LOG_LEVEL", "INFO"),
                    format="%(asctime)s [%(levelname)s] %(name)s - %(message)s", stream=sys.stdout)
log = logging.getLogger(STRATEGY_NAME)
load_dotenv(find_dotenv(usecwd=True))
API_KEY  = os.getenv("OPENALGO_API_KEY", "")
API_HOST = os.getenv("HOST_SERVER") or os.getenv("OPENALGO_HOST", "http://127.0.0.1:5000")
WS_URL   = os.getenv("WEBSOCKET_URL") or (
    f"ws://{os.getenv('WEBSOCKET_HOST','127.0.0.1')}:{os.getenv('WEBSOCKET_PORT','8765')}")
COSTS = cost_lookup(PRODUCT, EXCHANGE)


def signals_backtest(df):
    """Backtest entries on band breach. Bars where high > upper band -> long."""
    ind = get_indicators(INDICATOR_LIB)
    atr = ind.atr(df["high"], df["low"], df["close"], ATR_PERIOD)
    upper = df["close"].shift(1) + ATR_MULTIPLIER * atr.shift(1)
    lower = df["close"].shift(1) - ATR_MULTIPLIER * atr.shift(1)

    long_entry  = (df["high"] > upper).fillna(False).astype(bool)
    short_entry = (df["low"]  < lower).fillna(False).astype(bool)
    # Time-based exit: end of session
    is_eod = df.index.time >= pd.Timestamp("15:15").time()
    exits = pd.Series(is_eod, index=df.index)
    return long_entry, short_entry, exits


def run_backtest():
    import vectorbt as vbt
    log.info("BACKTEST: %s %s %s @ %s", STRATEGY_NAME, SYMBOL, EXCHANGE, INTERVAL)
    log.info("\n%s", format_cost_report(COSTS, INIT_CASH))
    client = api(api_key=API_KEY, host=API_HOST)
    end = datetime.now().date(); start = end - timedelta(days=LOOKBACK_DAYS)
    df = fetch_backtest_data(client, SYMBOL, EXCHANGE, INTERVAL,
                             start.strftime("%Y-%m-%d"), end.strftime("%Y-%m-%d"),
                             source=DATA_SOURCE)
    if df is None or len(df) < ATR_PERIOD * 3:
        log.error("Insufficient bars"); return
    long_e, short_e, exits = signals_backtest(df)
    size_pct = fixed_fractional_size(RISK_PER_TRADE, RISK.sl_pct, MAX_SIZE_PCT)
    log.info("Sizing: %.2f%% per trade", size_pct*100)
    pf = vbt.Portfolio.from_signals(
        df["close"],
        entries=long_e, short_entries=short_e,
        exits=exits, short_exits=exits,
        price=df["open"].shift(-1),
        init_cash=INIT_CASH, fees=COSTS.fees, fixed_fees=COSTS.fixed_fees,
        slippage=COSTS.slippage, size=size_pct, size_type="percent",
        sl_stop=RISK.sl_pct, tp_stop=RISK.tp_pct,
        sl_trail=False if RISK.trail_pct is None else RISK.trail_pct,
        freq=_freq(INTERVAL), min_size=LOT_SIZE, size_granularity=LOT_SIZE,
    )
    log.info("\n=== Stats ===\n%s", pf.stats())
    out = Path("backtests") / STRATEGY_NAME; out.mkdir(parents=True, exist_ok=True)
    pf.trades.records_readable.to_csv(out / f"{SYMBOL}_trades.csv", index=False)


def _freq(i):
    return {"1m":"1min","3m":"3min","5m":"5min","10m":"10min","15m":"15min",
            "30m":"30min","1h":"1H","D":"1D"}.get(i, "5min")


# ============================================================================
# LIVE: tick-driven LIMIT placement + modification
# ============================================================================

def run_live():
    log.info("LIVE: %s %s @ %s (real-time LIMIT)", STRATEGY_NAME, SYMBOL, INTERVAL)
    client = api(api_key=API_KEY, host=API_HOST, ws_url=WS_URL)
    try:
        run_preflight(client, symbol=SYMBOL, exchange=EXCHANGE, expected_exchange_env=EXCHANGE)
    except Exception as e:
        log.error("Preflight failed: %s - aborting", e); return

    state_db = StrategyState(_HERE / "state.db")
    qty = compute_live_qty(client, SYMBOL, EXCHANGE, sl_pct=RISK.sl_pct,
                           risk_per_trade=RISK_PER_TRADE,
                           lot_size=LOT_SIZE, min_qty=LOT_SIZE,
                           max_capital_pct=MAX_SIZE_PCT)
    if qty <= 0:
        log.error("qty=0 - aborting"); state_db.close(); return
    log.info("Live qty: %d", qty)

    client.connect()
    slip = SlippageTracker(assumed_pct=COSTS.slippage)

    # Compute initial bands from warmup history
    ind = get_indicators(INDICATOR_LIB)
    df = warmup_live_data(client, SYMBOL, EXCHANGE, INTERVAL, lookback_bars=300,
                          source=DATA_SOURCE)
    atr_series = ind.atr(df["high"], df["low"], df["close"], ATR_PERIOD)
    last_close = float(df["close"].iloc[-1])
    last_atr   = float(atr_series.iloc[-1])
    upper = last_close + ATR_MULTIPLIER * last_atr
    lower = last_close - ATR_MULTIPLIER * last_atr
    log.info("Initial bands: upper=%.2f lower=%.2f (close=%.2f atr=%.2f)",
             upper, lower, last_close, last_atr)

    state = {
        "buy_oid": None, "sell_oid": None,
        "buy_price": None, "sell_price": None,
        "last_modify": 0.0,
        "position": None,
    }

    risk_mgr = RiskManager(client, STRATEGY_NAME, RISK,
                           on_exit_callback=lambda *a: state.update({"position": None}),
                           slippage_tracker=slip, state=state_db)
    state["last_band_update"] = time.time()      # for MARKET fallback decision

    def place_limit(action, price):
        try:
            r = client.placeorder(
                strategy=STRATEGY_NAME, symbol=SYMBOL, exchange=EXCHANGE,
                action=action, price_type="LIMIT", product=PRODUCT,
                quantity=qty, price=str(round(price, 2)),
            )
            return r.get("orderid") if isinstance(r, dict) else None
        except Exception:
            log.exception("place_limit failed for %s @ %.2f", action, price); return None

    def modify_limit(oid, action, price):
        if oid is None: return
        try:
            client.modifyorder(
                order_id=oid, strategy=STRATEGY_NAME,
                symbol=SYMBOL, exchange=EXCHANGE,
                action=action, price_type="LIMIT", product=PRODUCT,
                quantity=qty, price=str(round(price, 2)),
            )
        except Exception:
            log.exception("modifyorder failed for %s @ %.2f", action, price)

    # Place initial LIMIT orders pegged at bands +/- offset
    state["buy_price"]  = round(upper * (1.0 + LIMIT_OFFSET_PCT), 2)
    state["sell_price"] = round(lower * (1.0 - LIMIT_OFFSET_PCT), 2)
    state["buy_oid"]  = place_limit("BUY",  state["buy_price"])
    state["sell_oid"] = place_limit("SELL", state["sell_price"])
    log.info("LIMIT BUY @ %.2f oid=%s | LIMIT SELL @ %.2f oid=%s",
             state["buy_price"], state["buy_oid"], state["sell_price"], state["sell_oid"])

    def on_tick(data):
        if state["position"] is not None:
            return  # don't modify after fill
        try:
            ltp = float(data.get("data", {}).get("ltp", 0))
        except (TypeError, ValueError):
            return
        if ltp <= 0: return

        # Throttle modifications
        if time.time() - state["last_modify"] < LIMIT_MODIFY_THROTTLE:
            return

        # Recompute bands using current LTP as proxy for live close
        new_upper = ltp + ATR_MULTIPLIER * last_atr
        new_lower = ltp - ATR_MULTIPLIER * last_atr
        new_buy  = round(new_upper * (1.0 + LIMIT_OFFSET_PCT), 2)
        new_sell = round(new_lower * (1.0 - LIMIT_OFFSET_PCT), 2)

        # Modify if drift is more than 1 tick
        if state["buy_oid"] and abs(new_buy - state["buy_price"]) >= 0.05:
            log.debug("Modify BUY %.2f -> %.2f", state["buy_price"], new_buy)
            modify_limit(state["buy_oid"], "BUY", new_buy)
            state["buy_price"] = new_buy
        if state["sell_oid"] and abs(new_sell - state["sell_price"]) >= 0.05:
            log.debug("Modify SELL %.2f -> %.2f", state["sell_price"], new_sell)
            modify_limit(state["sell_oid"], "SELL", new_sell)
            state["sell_price"] = new_sell
        state["last_modify"] = time.time()

    instruments = [{"exchange": EXCHANGE, "symbol": SYMBOL}]
    client.subscribe_ltp(instruments, on_data_received=on_tick)

    # OCO + risk handoff polling
    while not stop_event.is_set():
        now = datetime.now()
        if now.hour > SQUARE_OFF_TIME_HOUR or (
            now.hour == SQUARE_OFF_TIME_HOUR and now.minute >= SQUARE_OFF_TIME_MIN):
            log.info("Square-off")
            try: client.cancelallorder(strategy=STRATEGY_NAME)
            except Exception: log.exception("cancelallorder failed")
            try: client.closeposition(strategy=STRATEGY_NAME)
            except Exception: log.exception("closeposition failed")
            break

        # Check OCO fill
        if state["position"] is None:
            for side, oid_key, price_key in [("BUY","buy_oid","buy_price"),
                                              ("SELL","sell_oid","sell_price")]:
                oid = state.get(oid_key)
                if not oid: continue
                try:
                    r = client.orderstatus(order_id=oid, strategy=STRATEGY_NAME)
                    d = r.get("data", {}) if isinstance(r, dict) else {}
                    if d.get("order_status") == "complete":
                        fill = float(d.get("average_price") or d.get("price") or 0)
                        slip.record(state[price_key], fill, qty, side)
                        log.info("%s LIMIT filled @ %.2f", side, fill)
                        other = "sell_oid" if oid_key == "buy_oid" else "buy_oid"
                        if state[other]:
                            try:
                                client.cancelorder(order_id=state[other], strategy=STRATEGY_NAME)
                                log.info("Cancelled OCO leg")
                            except Exception:
                                log.exception("OCO cancel failed")
                        pos = Position(SYMBOL, EXCHANGE, side, qty, fill,
                                       time.time(), PRODUCT, STRATEGY_NAME)
                        state["position"] = pos
                        risk_mgr.set_position(pos)
                        break
                except Exception:
                    log.exception("orderstatus poll failed")

        # LIMIT timeout MARKET fallback: if we're past LIMIT_TIMEOUT_SEC and
        # the price has moved through one of the bands without filling, cancel
        # the LIMIT and place MARKET to guarantee execution. The "price moved
        # through" check uses get_ltp() against the band level.
        if state["position"] is None and LIMIT_TIMEOUT_SEC > 0:
            try:
                snapshot = client.get_ltp() or {}
                ltp_now = float(
                    snapshot.get(EXCHANGE, {}).get(SYMBOL, {}).get("ltp", 0) or 0
                )
            except Exception:
                ltp_now = 0
            if ltp_now > 0 and (time.time() - state["last_band_update"]) > LIMIT_TIMEOUT_SEC:
                fallback_side = None
                if state["buy_price"] and ltp_now >= state["buy_price"]:
                    fallback_side = "BUY"
                elif state["sell_price"] and ltp_now <= state["sell_price"]:
                    fallback_side = "SELL"
                if fallback_side is not None:
                    log.warning("LIMIT timeout: price %.2f past %s band - falling back to MARKET",
                                ltp_now, fallback_side)
                    # Cancel both LIMIT orders
                    for o in (state["buy_oid"], state["sell_oid"]):
                        if o:
                            try: client.cancelorder(order_id=o, strategy=STRATEGY_NAME)
                            except Exception: log.exception("cancel before MARKET failed")
                    # MARKET entry
                    try:
                        r = client.placeorder(
                            strategy=STRATEGY_NAME, symbol=SYMBOL, exchange=EXCHANGE,
                            action=fallback_side, price_type="MARKET",
                            product=PRODUCT, quantity=qty,
                        )
                        oid = r.get("orderid") if isinstance(r, dict) else None
                        if oid:
                            d = client.orderstatus(order_id=oid, strategy=STRATEGY_NAME)
                            fill = float(d.get("data", {}).get("average_price")
                                         or d.get("data", {}).get("price") or ltp_now)
                            decided = state["buy_price"] if fallback_side == "BUY" else state["sell_price"]
                            slip.record(decided, fill, qty, fallback_side)
                            pos = Position(SYMBOL, EXCHANGE, fallback_side, qty, fill,
                                           time.time(), PRODUCT, STRATEGY_NAME)
                            state["position"] = pos
                            risk_mgr.set_position(pos)
                            log.info("MARKET fallback filled %s @ %.2f", fallback_side, fill)
                    except Exception:
                        log.exception("MARKET fallback failed")

        if state["position"] is not None and state["position"].closed:
            log.info("Position closed by risk manager - end of run")
            break

        stop_event.wait(1)

    try: client.unsubscribe_ltp(instruments)
    except Exception: pass
    risk_mgr.stop()
    try: client.disconnect()
    except Exception: pass
    state_db.close()
    log.info("\n%s", slip.report())


stop_event = threading.Event()
def _sh(s, f): log.info("signal %d - shutting down", s); stop_event.set()
signal.signal(signal.SIGTERM, _sh); signal.signal(signal.SIGINT, _sh)


def main():
    p = argparse.ArgumentParser()
    p.add_argument("--mode", choices=["backtest","live"], default=os.getenv("MODE","live"))
    a = p.parse_args()
    run_backtest() if a.mode == "backtest" else run_live()

if __name__ == "__main__": main()

"""
Bollinger Band Squeeze + Breakout - dual-mode strategy.

Squeeze: BBWidth < its rolling minimum over LOOKBACK bars.
After squeeze, enter long when close breaks above upper band, short below lower.
"""
import argparse, logging, os, signal, sys, threading, time
from datetime import datetime, timedelta
from pathlib import Path

import pandas as pd
from dotenv import find_dotenv, load_dotenv

_HERE = Path(__file__).resolve().parent
for parent in [_HERE, *_HERE.parents]:
    candidate = parent / ".claude" / "skills" / "algo-expert" / "rules" / "assets" / "core"
    if candidate.exists():
        sys.path.insert(0, str(candidate.parent)); break

from openalgo import api  # noqa: E402
from core.cost_model import lookup as cost_lookup, format_cost_report, SlippageTracker  # noqa: E402
from core.indicator_adapter import get_indicators  # noqa: E402
from core.data_router import fetch_backtest_data, warmup_live_data, BarCloseWatcher  # noqa: E402
from core.risk_manager import RiskManager, RiskConfig, Position  # noqa: E402
from core.sizing import fixed_fractional_size, compute_live_qty  # noqa: E402
from core.preflight import run_preflight, find_existing_open_position  # noqa: E402
from core.state import StrategyState, reconcile_with_broker  # noqa: E402

# === Config ===
SYMBOL          = "TCS"
EXCHANGE        = os.getenv("OPENALGO_STRATEGY_EXCHANGE", os.getenv("EXCHANGE", "NSE"))
INTERVAL        = "15m"
PRODUCT         = "MIS"
LOT_SIZE        = 1
STRATEGY_NAME   = os.getenv("STRATEGY_NAME", "bb_squeeze")
DATA_SOURCE     = os.getenv("DATA_SOURCE", "api")
RISK_PER_TRADE  = 0.005
MAX_SIZE_PCT    = 0.50
BB_PERIOD       = 20
BB_STD          = 2.0
SQUEEZE_LOOKBACK = 50
INDICATOR_LIB   = "openalgo"
EXECUTION_TYPE  = "eoc"
POLL_INTERVAL_SEC = 15
RISK = RiskConfig(sl_pct=0.012, tp_pct=0.025, trail_pct=0.01, time_exit_min=180)
INIT_CASH       = 1_000_000
LOOKBACK_DAYS   = 365 * 2

logging.basicConfig(level=os.getenv("LOG_LEVEL", "INFO"),
                    format="%(asctime)s [%(levelname)s] %(name)s - %(message)s", stream=sys.stdout)
log = logging.getLogger(STRATEGY_NAME)
load_dotenv(find_dotenv(usecwd=True))
API_KEY  = os.getenv("OPENALGO_API_KEY", "")
API_HOST = os.getenv("HOST_SERVER") or os.getenv("OPENALGO_HOST", "http://127.0.0.1:5000")
WS_URL   = os.getenv("WEBSOCKET_URL") or (
    f"ws://{os.getenv('WEBSOCKET_HOST','127.0.0.1')}:{os.getenv('WEBSOCKET_PORT','8765')}")
COSTS = cost_lookup(PRODUCT, EXCHANGE)


def signals(df):
    ind = get_indicators(INDICATOR_LIB)
    upper, mid, lower = ind.bbands(df["close"], BB_PERIOD, BB_STD)
    width = (upper - lower) / mid
    width_min = width.rolling(SQUEEZE_LOOKBACK, min_periods=SQUEEZE_LOOKBACK // 2).min()
    in_squeeze = (width <= width_min * 1.05).fillna(False)
    was_squeeze = in_squeeze.shift(1).fillna(False)

    long_break  = (df["close"] > upper) & was_squeeze
    short_break = (df["close"] < lower) & was_squeeze
    entries = long_break.fillna(False).astype(bool)
    exits   = short_break.fillna(False).astype(bool)
    return entries, exits


def run_backtest():
    import vectorbt as vbt
    log.info("BACKTEST: %s %s %s @ %s", STRATEGY_NAME, SYMBOL, EXCHANGE, INTERVAL)
    log.info("\n%s", format_cost_report(COSTS, INIT_CASH))
    client = api(api_key=API_KEY, host=API_HOST)
    end = datetime.now().date(); start = end - timedelta(days=LOOKBACK_DAYS)
    df = fetch_backtest_data(client, SYMBOL, EXCHANGE, INTERVAL,
                             start.strftime("%Y-%m-%d"), end.strftime("%Y-%m-%d"),
                             source=DATA_SOURCE)
    if df is None or len(df) < SQUEEZE_LOOKBACK * 3:
        log.error("Insufficient bars"); return
    entries, exits = signals(df)
    size_pct = fixed_fractional_size(RISK_PER_TRADE, RISK.sl_pct, MAX_SIZE_PCT)
    log.info("Sizing: %.2f%% per trade", size_pct*100)
    pf = vbt.Portfolio.from_signals(
        df["close"], entries=entries, exits=exits,
        price=df["open"].shift(-1),
        init_cash=INIT_CASH, fees=COSTS.fees, fixed_fees=COSTS.fixed_fees,
        slippage=COSTS.slippage, size=size_pct, size_type="percent",
        sl_stop=RISK.sl_pct, tp_stop=RISK.tp_pct,
        sl_trail=False if RISK.trail_pct is None else RISK.trail_pct,
        freq=_freq(INTERVAL), min_size=LOT_SIZE, size_granularity=LOT_SIZE,
    )
    log.info("\n=== Stats ===\n%s", pf.stats())
    out = Path("backtests") / STRATEGY_NAME; out.mkdir(parents=True, exist_ok=True)
    pf.trades.records_readable.to_csv(out / f"{SYMBOL}_trades.csv", index=False)


def _freq(i):
    return {"1m":"1min","3m":"3min","5m":"5min","10m":"10min","15m":"15min",
            "30m":"30min","1h":"1H","D":"1D"}.get(i, "5min")


def run_live():
    log.info("LIVE: %s %s @ %s", STRATEGY_NAME, SYMBOL, INTERVAL)
    client = api(api_key=API_KEY, host=API_HOST, ws_url=WS_URL)
    try:
        run_preflight(client, symbol=SYMBOL, exchange=EXCHANGE, expected_exchange_env=EXCHANGE)
    except Exception as e:
        log.error("Preflight failed: %s - aborting", e); return

    state_db = StrategyState(_HERE / "state.db")
    qty = compute_live_qty(client, SYMBOL, EXCHANGE, sl_pct=RISK.sl_pct,
                           risk_per_trade=RISK_PER_TRADE,
                           lot_size=LOT_SIZE, min_qty=LOT_SIZE,
                           max_capital_pct=MAX_SIZE_PCT)
    if qty <= 0:
        log.error("qty=0 - aborting"); state_db.close(); return
    log.info("Live qty: %d", qty)

    client.connect()
    slip = SlippageTracker(assumed_pct=COSTS.slippage)
    state = {"position": None}
    risk_mgr = RiskManager(client, STRATEGY_NAME, RISK,
                           on_exit_callback=lambda *a: state.update({"position": None}),
                           slippage_tracker=slip, state=state_db)
    resumed = reconcile_with_broker(state_db, client, SYMBOL, EXCHANGE)
    if resumed is not None:
        pos = Position(resumed.symbol, resumed.exchange, resumed.side, resumed.qty,
                       resumed.entry_price, resumed.entry_time, resumed.product, STRATEGY_NAME)
        state["position"] = pos
        risk_mgr.set_position(pos, restore_watermark=resumed.watermark)
    warmup_live_data(client, SYMBOL, EXCHANGE, INTERVAL, source=DATA_SOURCE)


    def on_bar_close(df):
        entries, exits = signals(df)
        if len(df) < 3: return
        ltp = float(df["close"].iloc[-2])
        bar_ts = str(df.index[-2])
        if state_db.signal_already_acted(STRATEGY_NAME, bar_ts): return
        if entries.iloc[-2] and state["position"] is None:
            if find_existing_open_position(client, SYMBOL, EXCHANGE) is not None:
                log.warning("Broker has open pos - skip ENTRY")
                state_db.mark_signal_acted(STRATEGY_NAME, bar_ts); return
            log.info("ENTRY (squeeze break) %s @ %.2f", df.index[-2], ltp)
            r = client.placeorder(strategy=STRATEGY_NAME, symbol=SYMBOL, exchange=EXCHANGE,
                                  action="BUY", price_type="MARKET", product=PRODUCT, quantity=qty)
            oid = r.get("orderid") if isinstance(r, dict) else None
            fill = _wait_fill(client, oid, ltp) if oid else ltp
            slip.record(ltp, fill, qty, "BUY")
            pos = Position(SYMBOL, EXCHANGE, "BUY", qty, fill, time.time(), PRODUCT, STRATEGY_NAME)
            state["position"] = pos; risk_mgr.set_position(pos)
            state_db.mark_signal_acted(STRATEGY_NAME, bar_ts)
        elif exits.iloc[-2] and state["position"] is not None:
            log.info("EXIT %s @ %.2f", df.index[-2], ltp)
            try:
                client.placesmartorder(strategy=STRATEGY_NAME, symbol=SYMBOL, exchange=EXCHANGE,
                                       action="SELL", price_type="MARKET", product=PRODUCT,
                                       quantity=state["position"].qty, position_size=0)
            except Exception: log.exception("exit failed")
            risk_mgr.clear_position(); state["position"] = None
            state_db.mark_signal_acted(STRATEGY_NAME, bar_ts)

    watcher = BarCloseWatcher(client, SYMBOL, EXCHANGE, INTERVAL, on_bar_close,
                              poll_interval_sec=POLL_INTERVAL_SEC, stop_event=stop_event)
    try: watcher.run()
    finally:
        risk_mgr.stop()
        try: client.disconnect()
        except Exception: pass
        state_db.close()
        log.info("\n%s", slip.report())


def _wait_fill(client, oid, fallback, retries=10, sleep_s=0.5):
    for _ in range(retries):
        try:
            r = client.orderstatus(order_id=oid, strategy=STRATEGY_NAME)
            d = r.get("data", {}) if isinstance(r, dict) else {}
            if d.get("order_status") == "complete":
                avg = d.get("average_price") or d.get("price")
                if avg: return float(avg)
        except Exception: log.exception("orderstatus poll failed")
        time.sleep(sleep_s)
    return fallback


stop_event = threading.Event()
def _sh(s, f): log.info("signal %d - shutting down", s); stop_event.set()
signal.signal(signal.SIGTERM, _sh); signal.signal(signal.SIGINT, _sh)


def main():
    p = argparse.ArgumentParser()
    p.add_argument("--mode", choices=["backtest","live"], default=os.getenv("MODE","live"))
    a = p.parse_args()
    run_backtest() if a.mode == "backtest" else run_live()

if __name__ == "__main__": main()

"""
cost_model.py - Real-world transaction cost and slippage model for Indian markets.

Centralizes the 4-segment Indian fee structure (Intraday/Delivery Equity, F&O Futures,
F&O Options) and per-segment slippage assumptions. These same constants flow into:

  - VectorBT backtests via fees / fixed_fees / slippage parameters
  - Live runners via LIMIT-with-offset placement (slippage protection)
  - Drift reports comparing assumed vs measured slippage

Broker-neutral: override individual constants for your broker's actual rates.

Reference: matches conventions in vectorbt-backtesting-skills/indian-market-costs.md
"""
from dataclasses import dataclass


# ---------------------------------------------------------------------------
# Segment fee table (Indian Market Standard)
#
# Derived from STT + exchange transaction + GST + SEBI + stamp duty across a
# Rs 10L turnover. Brokerage is conservative Rs 20 per order across the board.
# Adjust to your actual broker rates if needed.
# ---------------------------------------------------------------------------

@dataclass(frozen=True)
class CostBlock:
    fees: float          # decimal, applied to turnover per side (0.001 = 0.1%)
    fixed_fees: float    # absolute, applied per order
    slippage: float      # decimal, applied to fill price per side
    label: str

INTRADAY_EQ = CostBlock(
    fees=0.000225,       # 0.0225% per side (statutory only, MIS pays no STT on buy)
    fixed_fees=20.0,     # Rs 20 brokerage per order
    slippage=0.0005,     # 5 bps - liquid intraday equity
    label="Intraday Equity (MIS)",
)

DELIVERY_EQ = CostBlock(
    fees=0.00111,        # 0.111% per side (STT 0.1% on both + statutory)
    fixed_fees=20.0,     # Conservative; many brokers offer free delivery
    slippage=0.0003,     # 3 bps - daily-bar delivery, less time-sensitive
    label="Delivery Equity (CNC)",
)

FUT_NRML = CostBlock(
    fees=0.00018,        # 0.018% per side (STT 0.02% sell side + statutory)
    fixed_fees=20.0,
    slippage=0.0002,     # 2 bps - very liquid index futures
    label="F&O Futures (NRML)",
)

OPT_NRML = CostBlock(
    fees=0.00098,        # 0.098% per side (STT 0.1% sell side + statutory)
    fixed_fees=20.0,
    slippage=0.0010,     # 10 bps - wider option spreads
    label="F&O Options (NRML)",
)

ILLIQUID_FALLBACK = CostBlock(
    fees=0.00111,
    fixed_fees=20.0,
    slippage=0.0030,     # 30 bps - thin names
    label="Illiquid (manual override)",
)


# ---------------------------------------------------------------------------
# Lookup helpers
# ---------------------------------------------------------------------------

# Map (product, exchange-or-instrument) -> CostBlock.
# Falls back conservatively if the pair isn't found.
_TABLE = {
    ("MIS", "NSE"):   INTRADAY_EQ,
    ("MIS", "BSE"):   INTRADAY_EQ,
    ("CNC", "NSE"):   DELIVERY_EQ,
    ("CNC", "BSE"):   DELIVERY_EQ,
    ("NRML", "NFO"):  FUT_NRML,    # default to futures; override for options below
    ("NRML", "BFO"):  FUT_NRML,
    ("NRML", "MCX"):  FUT_NRML,
    ("NRML", "CDS"):  FUT_NRML,
    ("NRML", "BCD"):  FUT_NRML,
}


def lookup(product, exchange, instrument_type=None):
    """
    Resolve a CostBlock for a (product, exchange) pair.

    instrument_type: optional. Pass "OPT" to force the option-fee block when
    the exchange is a derivatives one (NFO/BFO).
    """
    if instrument_type and instrument_type.upper().startswith("OPT"):
        if exchange in ("NFO", "BFO"):
            return OPT_NRML
    return _TABLE.get((product.upper(), exchange.upper()), DELIVERY_EQ)


# ---------------------------------------------------------------------------
# Reporting helpers
# ---------------------------------------------------------------------------

def cost_summary(block, turnover):
    """
    Estimate one-side cost on a given rupee turnover. Useful for printing
    backtest cost assumptions before running.
    """
    statutory = turnover * block.fees
    slip      = turnover * block.slippage
    brokerage = block.fixed_fees
    return {
        "label": block.label,
        "turnover": turnover,
        "statutory_per_side": statutory,
        "slippage_per_side": slip,
        "brokerage_per_order": brokerage,
        "round_trip": (statutory + slip) * 2 + brokerage * 2,
    }


def format_cost_report(block, turnover):
    s = cost_summary(block, turnover)
    return (
        f"=== Cost Model: {s['label']} ===\n"
        f"  Statutory + Exchange:  {block.fees*100:.4f}% per side\n"
        f"  Brokerage:             Rs {block.fixed_fees:.0f} per order\n"
        f"  Slippage:              {block.slippage*100:.4f}% per side\n"
        f"  On Rs {s['turnover']:,.0f} turnover:\n"
        f"    statutory  = Rs {s['statutory_per_side']:,.2f} (per side)\n"
        f"    slippage   = Rs {s['slippage_per_side']:,.2f} (per side)\n"
        f"    brokerage  = Rs {s['brokerage_per_order']:,.0f} (per order)\n"
        f"    round-trip = Rs {s['round_trip']:,.2f}"
    )


# ---------------------------------------------------------------------------
# Live-mode slippage tracker
# ---------------------------------------------------------------------------

class SlippageTracker:
    """
    Records realized slippage per fill and produces an end-of-session report.

    Usage:
        tracker = SlippageTracker(assumed_pct=0.0005)
        tracker.record(decision_price=100.0, fill_price=100.05)
        ...
        print(tracker.report())
    """

    def __init__(self, assumed_pct):
        self.assumed_pct = assumed_pct
        self.fills = []

    def record(self, decision_price, fill_price, qty=1, side="BUY"):
        if decision_price <= 0:
            return
        signed = (fill_price - decision_price) if side == "BUY" else (decision_price - fill_price)
        self.fills.append({
            "decision_price": decision_price,
            "fill_price": fill_price,
            "qty": qty,
            "side": side,
            "signed_slip_abs": signed,
            "signed_slip_pct": signed / decision_price,
        })

    def measured_pct(self):
        if not self.fills:
            return 0.0
        return sum(f["signed_slip_pct"] for f in self.fills) / len(self.fills)

    def report(self):
        if not self.fills:
            return "Slippage: no fills recorded."
        m = self.measured_pct()
        a = self.assumed_pct
        ratio = (m / a) if a else float("inf")
        worst = max(self.fills, key=lambda f: abs(f["signed_slip_pct"]))
        drift = "OK" if abs(ratio) < 2.0 else "DRIFT - measured >2x assumed"
        return (
            f"Slippage Report:\n"
            f"  Fills:               {len(self.fills)}\n"
            f"  Assumed (per side):  {a*100:.4f}%\n"
            f"  Measured (avg):      {m*100:.4f}%\n"
            f"  Ratio measured/assumed: {ratio:.2f}x  [{drift}]\n"
            f"  Worst slip: {worst['signed_slip_pct']*100:.4f}% "
            f"({worst['side']} @ decided {worst['decision_price']:.2f}, "
            f"filled {worst['fill_price']:.2f})"
        )

"""
data_router.py - Unified data access for backtest and live modes.

Three data sources supported:
  1. OpenAlgo API           - source="api"     (live broker fetch)
  2. OpenAlgo Historify     - source="db"      (SDK reads stored DuckDB)
  3. Direct DuckDB           - source="duckdb:/path/to/file.duckdb"
                                (auto-detects Historify vs custom format)

Backtest mode:
  - Single-shot fetch via fetch_backtest_data()
  - Returns a normalized DataFrame (datetime index, lowercase OHLCV columns)

Live mode:
  - warmup_live_data()  - last N bars so indicators are valid on bar 1
  - BarCloseWatcher     - polls history(), fires callback once per closed bar
                          (uses iloc[-2] - iloc[-1] is the forming bar)

WebSocket reconnection:
  - reconnect_ws()       - retry loop wrapping client.connect / subscribe
"""
from datetime import datetime, timedelta
import logging
import time

import pandas as pd

log = logging.getLogger(__name__)


# ---------------------------------------------------------------------------
# Normalization
# ---------------------------------------------------------------------------

def normalize_history(df):
    """Coerce data into a sorted DatetimeIndex DataFrame with lowercase OHLCV columns."""
    if df is None or len(df) == 0:
        return df
    df = df.copy()
    if "timestamp" in df.columns:
        df["timestamp"] = pd.to_datetime(df["timestamp"])
        df = df.set_index("timestamp")
    else:
        df.index = pd.to_datetime(df.index)
    df = df.sort_index()
    df.columns = [c.lower() for c in df.columns]
    return df


# ---------------------------------------------------------------------------
# Source dispatch
# ---------------------------------------------------------------------------

def fetch_backtest_data(client, symbol, exchange, interval,
                        start_date, end_date, source="api"):
    """
    Single-shot fetch for backtest mode. Routes to API, Historify (via SDK),
    or direct DuckDB based on `source`.

    source can be:
      - "api"           OpenAlgo live broker fetch
      - "db"            OpenAlgo Historify via SDK (client.history(source="db"))
      - "duckdb:/path"  direct DuckDB file read (skips OpenAlgo entirely)
    """
    log.info("Fetching backtest data: %s %s %s %s..%s (source=%s)",
             symbol, exchange, interval, start_date, end_date, source)

    if source.startswith("duckdb:"):
        return fetch_from_duckdb(
            db_path=source[len("duckdb:"):],
            symbol=symbol, exchange=exchange, interval=interval,
            start_date=start_date, end_date=end_date,
        )

    df = client.history(
        symbol=symbol, exchange=exchange, interval=interval,
        start_date=start_date, end_date=end_date,
        source=source,                    # "api" or "db"
    )
    df = normalize_history(df)
    if df is not None and len(df) > 0:
        log.info("Loaded %d bars from %s to %s", len(df), df.index[0], df.index[-1])
    return df


# ---------------------------------------------------------------------------
# Direct DuckDB readers (Historify auto-detection)
# ---------------------------------------------------------------------------

def fetch_from_duckdb(db_path, symbol, exchange, interval,
                      start_date=None, end_date=None):
    """
    Read OHLCV directly from a DuckDB file. Auto-detects Historify vs custom format.

    Historify schema:
        market_data(symbol, exchange, interval, timestamp(epoch), o,h,l,c,volume,oi)
    Custom OHLCV schema:
        ohlcv(symbol, date, time, open, high, low, close, volume)
    """
    try:
        import duckdb
    except ImportError as e:
        raise ImportError(
            "duckdb not installed - pip install duckdb to use source='duckdb:...'"
        ) from e

    fmt = _detect_duckdb_format(duckdb, db_path)
    if fmt == "historify":
        return _load_historify(duckdb, db_path, symbol, exchange, interval,
                               start_date, end_date)
    if fmt == "custom_ohlcv":
        return _load_custom_ohlcv(duckdb, db_path, symbol,
                                  start_date, end_date)
    raise ValueError(
        f"DuckDB at {db_path} has neither 'market_data' (Historify) "
        f"nor 'ohlcv' table - unable to auto-detect format. "
        f"Inspect with: duckdb.connect(db_path).execute('SHOW TABLES').fetchdf()"
    )


def _detect_duckdb_format(duckdb, db_path):
    con = duckdb.connect(db_path, read_only=True)
    try:
        tables = con.execute("SHOW TABLES").fetchdf()["name"].tolist()
        if "market_data" in tables:
            cols = con.execute("DESCRIBE market_data").fetchdf()["column_name"].tolist()
            if all(c in cols for c in ["symbol", "exchange", "interval", "timestamp"]):
                return "historify"
        if "ohlcv" in tables:
            return "custom_ohlcv"
        return "unknown"
    finally:
        con.close()


def _load_historify(duckdb, db_path, symbol, exchange, interval,
                    start_date=None, end_date=None):
    """
    Load from Historify. Storage intervals are only '1m' and 'D' - if the user
    asks for '5m', '15m' etc, we read '1m' and resample.
    """
    storage_interval = interval if interval in ("1m", "D") else "1m"
    where_clauses = ["symbol = ?", "exchange = ?", "interval = ?"]
    params = [symbol.upper(), exchange.upper(), storage_interval]

    if start_date:
        where_clauses.append("timestamp >= ?")
        params.append(int(pd.Timestamp(start_date).timestamp()))
    if end_date:
        where_clauses.append("timestamp <= ?")
        params.append(int(pd.Timestamp(end_date).timestamp() + 86400))  # inclusive end

    sql = (
        "SELECT timestamp, open, high, low, close, volume "
        "FROM market_data WHERE " + " AND ".join(where_clauses) +
        " ORDER BY timestamp"
    )
    con = duckdb.connect(db_path, read_only=True)
    try:
        df = con.execute(sql, params).fetchdf()
    finally:
        con.close()

    if df is None or len(df) == 0:
        log.warning("Historify returned 0 bars for %s/%s %s %s..%s",
                    symbol, exchange, interval, start_date, end_date)
        return df

    df["datetime"] = pd.to_datetime(df["timestamp"], unit="s")
    df = df.set_index("datetime").drop(columns=["timestamp"]).sort_index()

    if interval not in ("1m", "D") and storage_interval == "1m":
        df = resample_ohlcv(df, _interval_to_pandas(interval))

    log.info("Loaded %d bars from Historify (%s/%s %s)", len(df), symbol, exchange, interval)
    return df


def _load_custom_ohlcv(duckdb, db_path, symbol, start_date=None, end_date=None):
    where_clauses = ["symbol = ?"]
    params = [symbol]
    if start_date:
        where_clauses.append("date >= ?")
        params.append(str(start_date))
    if end_date:
        where_clauses.append("date <= ?")
        params.append(str(end_date))
    sql = (
        "SELECT date, time, open, high, low, close, volume FROM ohlcv "
        "WHERE " + " AND ".join(where_clauses) + " ORDER BY date, time"
    )
    con = duckdb.connect(db_path, read_only=True)
    try:
        df = con.execute(sql, params).fetchdf()
    finally:
        con.close()

    if df is None or len(df) == 0:
        return df

    df["datetime"] = pd.to_datetime(df["date"].astype(str) + " " + df["time"].astype(str))
    df = df.set_index("datetime").drop(columns=["date", "time"]).sort_index()
    log.info("Loaded %d bars from custom DuckDB (%s)", len(df), symbol)
    return df


def resample_ohlcv(df, timeframe="5min"):
    """Resample OHLCV with Indian market alignment (09:15 open)."""
    return df.resample(
        timeframe, origin="start_day", offset="9h15min",
        label="right", closed="right",
    ).agg({
        "open": "first", "high": "max", "low": "min",
        "close": "last", "volume": "sum",
    }).dropna()


def _interval_to_pandas(interval):
    return {
        "1m": "1min", "3m": "3min", "5m": "5min", "10m": "10min",
        "15m": "15min", "30m": "30min", "1h": "1H", "D": "1D",
    }.get(interval, "1min")


# ---------------------------------------------------------------------------
# Live warmup + bar-close watcher
# ---------------------------------------------------------------------------

def warmup_live_data(client, symbol, exchange, interval, lookback_bars=200,
                     source="api"):
    """Fetch enough history at startup so indicators are valid on bar 1."""
    end = datetime.now().date()
    if interval == "D":
        start = end - timedelta(days=lookback_bars * 2)
    elif interval in ("1h",):
        start = end - timedelta(days=max(60, lookback_bars // 6))
    else:
        start = end - timedelta(days=max(15, lookback_bars // 75))
    df = fetch_backtest_data(
        client, symbol, exchange, interval,
        start.strftime("%Y-%m-%d"), end.strftime("%Y-%m-%d"),
        source=source,
    )
    if df is None or len(df) < 50:
        log.warning("Warmup returned only %d bars - indicators may NaN initially.",
                    0 if df is None else len(df))
    return df


def poll_for_new_bar(client, symbol, exchange, interval, last_seen_ts,
                     lookback_days=7):
    """
    Re-fetch recent history. Returns (df, new_bar_closed).

    new_bar_closed is True ONLY when the most-recently-closed bar (iloc[-2]) is
    strictly newer than last_seen_ts. The first call (with last_seen_ts=None)
    seeds the timestamp WITHOUT firing a "new bar" event - prevents the
    off-by-one bug where startup re-evaluates an old signal as if it just fired.
    """
    end = datetime.now().date()
    start = end - timedelta(days=lookback_days)
    df = client.history(
        symbol=symbol, exchange=exchange, interval=interval,
        start_date=start.strftime("%Y-%m-%d"),
        end_date=end.strftime("%Y-%m-%d"),
        source="api",
    )
    df = normalize_history(df)
    if df is None or len(df) < 2:
        return df, False

    closed_ts = df.index[-2]

    if last_seen_ts is None:
        # First poll - just seed; do NOT fire callback (would re-evaluate
        # the most recent already-closed bar as if it were brand new)
        return df, False

    return df, closed_ts > last_seen_ts


class BarCloseWatcher:
    """
    Polls client.history() at POLL_INTERVAL_SEC, fires the on_bar_close
    callback once per newly-closed bar.

    On startup the first poll seeds last_seen_ts WITHOUT calling the callback.
    The callback fires only when a strictly-newer closed bar appears.
    """

    def __init__(self, client, symbol, exchange, interval,
                 on_bar_close, poll_interval_sec=15, lookback_days=7,
                 stop_event=None):
        self.client = client
        self.symbol = symbol
        self.exchange = exchange
        self.interval = interval
        self.on_bar_close = on_bar_close
        self.poll_interval_sec = poll_interval_sec
        self.lookback_days = lookback_days
        self.stop_event = stop_event
        self.last_seen_ts = None
        self._first_poll = True

    def run(self):
        while self.stop_event is None or not self.stop_event.is_set():
            try:
                df, is_new = poll_for_new_bar(
                    self.client, self.symbol, self.exchange, self.interval,
                    self.last_seen_ts, lookback_days=self.lookback_days,
                )
                if df is not None and len(df) >= 2:
                    if self._first_poll:
                        # Seed without firing - prevents stale-bar replay
                        self.last_seen_ts = df.index[-2]
                        self._first_poll = False
                        log.info("BarCloseWatcher: seeded last_seen_ts=%s "
                                 "(first poll, no signal fired)",
                                 self.last_seen_ts)
                    elif is_new:
                        self.last_seen_ts = df.index[-2]
                        try:
                            self.on_bar_close(df)
                        except Exception:
                            log.exception("on_bar_close raised - continuing")
            except Exception:
                log.exception("poll_for_new_bar failed - retrying after backoff")
                time.sleep(self.poll_interval_sec)
                continue

            if self.stop_event is not None:
                self.stop_event.wait(self.poll_interval_sec)
            else:
                time.sleep(self.poll_interval_sec)


# ---------------------------------------------------------------------------
# WebSocket reconnection wrapper
# ---------------------------------------------------------------------------

def reconnect_ws(client, instruments, on_data_received, mode="ltp",
                 stop_event=None, max_retries=None,
                 backoff_initial=2.0, backoff_max=30.0):
    """
    Connect to OpenAlgo WS and subscribe. On disconnect, retry with backoff.

    Loops until stop_event is set (or max_retries is hit). Use this in place
    of bare client.connect() + subscribe_*() to harden against network blips.
    """
    sub_fn = {
        "ltp":   client.subscribe_ltp,
        "quote": client.subscribe_quote,
        "depth": client.subscribe_depth,
    }.get(mode)
    if sub_fn is None:
        raise ValueError(f"Unknown mode: {mode}")

    backoff = backoff_initial
    attempts = 0
    while stop_event is None or not stop_event.is_set():
        try:
            client.connect()
            sub_fn(instruments, on_data_received=on_data_received)
            log.info("WS connected and subscribed (%s mode, %d instruments)",
                     mode, len(instruments))
            backoff = backoff_initial    # reset on successful connect
            # Block here until disconnected or stop requested
            while stop_event is None or not stop_event.is_set():
                if stop_event is not None:
                    stop_event.wait(1)
                else:
                    time.sleep(1)
            break
        except Exception:
            attempts += 1
            log.exception("WS connect/subscribe failed (attempt %d) - retrying in %.1fs",
                          attempts, backoff)
            if max_retries is not None and attempts >= max_retries:
                log.error("WS retries exhausted (%d) - giving up", max_retries)
                return
            time.sleep(backoff)
            backoff = min(backoff * 2, backoff_max)
        finally:
            try:
                if mode == "ltp":   client.unsubscribe_ltp(instruments)
                elif mode == "quote": client.unsubscribe_quote(instruments)
                elif mode == "depth": client.unsubscribe_depth(instruments)
            except Exception:
                pass
            try:
                client.disconnect()
            except Exception:
                pass

"""
indicator_adapter.py - Routes indicator calls to either openalgo.ta or talib.

Strategy templates pick a library at creation time. They import this module
and call ind.ema(...), ind.rsi(...), etc. Switching libraries is a one-line
change in the strategy file (LIBRARY = "openalgo" -> "talib").

talib doesn't have Supertrend, Donchian, Ichimoku, HMA, KAMA, ZLEMA, ALMA,
VWMA - those always route to openalgo.ta regardless of LIBRARY setting.
"""
import numpy as np
import pandas as pd


def _to_series(arr, index=None):
    if isinstance(arr, pd.Series):
        return arr
    return pd.Series(arr, index=index)


class _OpenAlgoBackend:
    """openalgo.ta backend (default). Fast Numba JIT, 100+ indicators."""

    name = "openalgo"

    def __init__(self):
        from openalgo import ta
        self._ta = ta

    def sma(self, close, period):
        return _to_series(self._ta.sma(close, period), getattr(close, "index", None))

    def ema(self, close, period):
        return _to_series(self._ta.ema(close, period), getattr(close, "index", None))

    def rsi(self, close, period=14):
        return _to_series(self._ta.rsi(close, period), getattr(close, "index", None))

    def macd(self, close, fast=12, slow=26, signal=9):
        macd, sig, hist = self._ta.macd(close, fast, slow, signal)
        idx = getattr(close, "index", None)
        return _to_series(macd, idx), _to_series(sig, idx), _to_series(hist, idx)

    def atr(self, high, low, close, period=14):
        return _to_series(self._ta.atr(high, low, close, period), getattr(close, "index", None))

    def bbands(self, close, period=20, std=2.0):
        upper, mid, lower = self._ta.bbands(close, period, std)
        idx = getattr(close, "index", None)
        return _to_series(upper, idx), _to_series(mid, idx), _to_series(lower, idx)

    def adx(self, high, low, close, period=14):
        return _to_series(self._ta.adx(high, low, close, period), getattr(close, "index", None))

    def stochastic(self, high, low, close, k=14, d=3, smooth=3):
        k_line, d_line = self._ta.stochastic(high, low, close, k, d, smooth)
        idx = getattr(close, "index", None)
        return _to_series(k_line, idx), _to_series(d_line, idx)

    def supertrend(self, high, low, close, period=10, multiplier=3.0):
        st, direction = self._ta.supertrend(high, low, close, period, multiplier)
        idx = getattr(close, "index", None)
        return _to_series(st, idx), _to_series(direction, idx)

    def donchian(self, high, low, period=20):
        upper, middle, lower = self._ta.donchian(high, low, period)
        idx = getattr(high, "index", None)
        return _to_series(upper, idx), _to_series(middle, idx), _to_series(lower, idx)

    def hma(self, close, period):
        return _to_series(self._ta.hma(close, period), getattr(close, "index", None))

    def kama(self, close, period=10):
        return _to_series(self._ta.kama(close, period), getattr(close, "index", None))

    def stdev(self, close, period):
        return _to_series(self._ta.stdev(close, period), getattr(close, "index", None))

    def crossover(self, a, b):
        return self._ta.crossover(a, b)

    def crossunder(self, a, b):
        return self._ta.crossunder(a, b)

    def exrem(self, primary, secondary):
        return self._ta.exrem(primary, secondary)


class _TaLibBackend:
    """talib backend. Standard library, faster than openalgo for some calls.
    Falls back to openalgo for indicators talib doesn't have."""

    name = "talib"

    def __init__(self):
        import talib
        self._tl = talib
        # Lazy fallback only when a missing indicator is requested
        self._fallback = None

    def _fb(self):
        if self._fallback is None:
            self._fallback = _OpenAlgoBackend()
        return self._fallback

    def _arr(self, x):
        if isinstance(x, pd.Series):
            return x.values.astype(np.float64)
        return np.asarray(x, dtype=np.float64)

    def sma(self, close, period):
        out = self._tl.SMA(self._arr(close), timeperiod=period)
        return _to_series(out, getattr(close, "index", None))

    def ema(self, close, period):
        out = self._tl.EMA(self._arr(close), timeperiod=period)
        return _to_series(out, getattr(close, "index", None))

    def rsi(self, close, period=14):
        out = self._tl.RSI(self._arr(close), timeperiod=period)
        return _to_series(out, getattr(close, "index", None))

    def macd(self, close, fast=12, slow=26, signal=9):
        macd, sig, hist = self._tl.MACD(
            self._arr(close), fastperiod=fast, slowperiod=slow, signalperiod=signal
        )
        idx = getattr(close, "index", None)
        return _to_series(macd, idx), _to_series(sig, idx), _to_series(hist, idx)

    def atr(self, high, low, close, period=14):
        out = self._tl.ATR(self._arr(high), self._arr(low), self._arr(close), timeperiod=period)
        return _to_series(out, getattr(close, "index", None))

    def bbands(self, close, period=20, std=2.0):
        upper, mid, lower = self._tl.BBANDS(
            self._arr(close), timeperiod=period, nbdevup=std, nbdevdn=std
        )
        idx = getattr(close, "index", None)
        return _to_series(upper, idx), _to_series(mid, idx), _to_series(lower, idx)

    def adx(self, high, low, close, period=14):
        out = self._tl.ADX(self._arr(high), self._arr(low), self._arr(close), timeperiod=period)
        return _to_series(out, getattr(close, "index", None))

    def stochastic(self, high, low, close, k=14, d=3, smooth=3):
        k_line, d_line = self._tl.STOCH(
            self._arr(high), self._arr(low), self._arr(close),
            fastk_period=k, slowk_period=smooth, slowd_period=d,
        )
        idx = getattr(close, "index", None)
        return _to_series(k_line, idx), _to_series(d_line, idx)

    def stdev(self, close, period):
        out = self._tl.STDDEV(self._arr(close), timeperiod=period, nbdev=1.0)
        return _to_series(out, getattr(close, "index", None))

    # talib doesn't have these - always route to openalgo
    def supertrend(self, *a, **kw):    return self._fb().supertrend(*a, **kw)
    def donchian(self, *a, **kw):       return self._fb().donchian(*a, **kw)
    def hma(self, *a, **kw):            return self._fb().hma(*a, **kw)
    def kama(self, *a, **kw):           return self._fb().kama(*a, **kw)
    def crossover(self, *a, **kw):      return self._fb().crossover(*a, **kw)
    def crossunder(self, *a, **kw):     return self._fb().crossunder(*a, **kw)
    def exrem(self, *a, **kw):          return self._fb().exrem(*a, **kw)


def get_indicators(library="openalgo"):
    """Return an indicator backend. library = 'openalgo' | 'talib'."""
    lib = library.lower().strip()
    if lib == "talib":
        return _TaLibBackend()
    return _OpenAlgoBackend()

"""
portfolio_runner.py - Multi-strategy supervisor with portfolio-level risk caps.

Reads a YAML config that lists strategies + caps, launches each strategy as a
subprocess, monitors aggregate P&L via OpenAlgo's tradebook/positionbook APIs,
and triggers a kill-switch when caps breach.

Backtest mode: aggregates per-strategy backtest equity curves and computes
the same caps post-hoc.

YAML schema:
    capital: 1000000
    portfolio_caps:
      portfolio_sl_pct: 0.02       # halt all strategies at -2% capital
      portfolio_tp_pct: 0.03       # halt all strategies at +3% capital
      daily_loss_pct: 0.015        # daily loss limit
      daily_target_pct: 0.025      # daily target
      max_concurrent_positions: 5
      max_symbol_concentration: 0.30   # one symbol cannot exceed 30% of capital
    strategies:
      - name: ema_sbin
        path: strategies/ema_sbin/strategy.py
      - name: rsi_reliance
        path: strategies/rsi_reliance/strategy.py
"""
from datetime import datetime
import logging
import os
import signal
import subprocess
import sys
import threading
import time
from pathlib import Path

import yaml


log = logging.getLogger(__name__)


# ---------------------------------------------------------------------------
# Config loader
# ---------------------------------------------------------------------------

def load_config(path):
    with open(path) as f:
        cfg = yaml.safe_load(f)
    return cfg


# ---------------------------------------------------------------------------
# Live-mode supervisor
# ---------------------------------------------------------------------------

class StrategyProcess:
    def __init__(self, name, script_path, env=None):
        self.name = name
        self.script_path = script_path
        self.env = env or {}
        self.proc = None

    def start(self, mode="live"):
        env = os.environ.copy()
        env.update(self.env)
        env["MODE"] = mode
        env["STRATEGY_NAME"] = self.name
        log.info("Starting strategy %s (%s)", self.name, self.script_path)
        self.proc = subprocess.Popen(
            [sys.executable, self.script_path],
            env=env,
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,
            text=True,
            bufsize=1,
        )

        # Pump stdout to log
        threading.Thread(target=self._pump, daemon=True).start()

    def _pump(self):
        if self.proc is None or self.proc.stdout is None:
            return
        for line in self.proc.stdout:
            log.info("[%s] %s", self.name, line.rstrip())

    def is_alive(self):
        return self.proc is not None and self.proc.poll() is None

    def stop(self, timeout=15):
        if not self.is_alive():
            return
        log.info("Stopping strategy %s (SIGTERM)", self.name)
        try:
            self.proc.terminate()
            self.proc.wait(timeout=timeout)
        except subprocess.TimeoutExpired:
            log.warning("Strategy %s did not stop in %ds, sending SIGKILL", self.name, timeout)
            self.proc.kill()


class PortfolioRunner:
    def __init__(self, config_path, client_factory, mode="live"):
        self.cfg = load_config(config_path)
        self.client_factory = client_factory
        self.mode = mode
        self.client = client_factory()
        self.children = [
            StrategyProcess(s["name"], s["path"], env=s.get("env"))
            for s in self.cfg.get("strategies", [])
        ]
        self.start_capital = float(self.cfg.get("capital", 1_000_000))
        self.caps = self.cfg.get("portfolio_caps", {})
        self.day_open_realized = 0.0    # realized P&L at start of day
        self.daily_reset_done = False
        self.kill_reason = None
        self._stop_event = threading.Event()

    # --- Lifecycle ----------------------------------------------------------

    def start(self):
        log.info("Portfolio runner: starting %d strategies (mode=%s)",
                 len(self.children), self.mode)
        log.info("Caps: %s", self.caps)
        for child in self.children:
            child.start(mode=self.mode)
        # Monitor loop
        threading.Thread(target=self._monitor, daemon=True).start()

    def stop_all(self, reason=""):
        if self.kill_reason is None and reason:
            self.kill_reason = reason
            log.warning("KILL SWITCH: %s", reason)
        for child in self.children:
            child.stop()
        # Defensive: cancel pending orders + close positions
        try:
            self.client.cancelallorder(strategy="PortfolioRunner")
        except Exception:
            log.exception("cancelallorder failed")
        try:
            self.client.closeposition(strategy="PortfolioRunner")
        except Exception:
            log.exception("closeposition failed")

    # --- Monitor loop -------------------------------------------------------

    def _monitor(self):
        while not self._stop_event.is_set():
            try:
                self._reset_at_midnight()
                pnl = self._compute_total_pnl()
                breach = self._check_caps(pnl)
                if breach:
                    self.stop_all(reason=breach)
                    return
            except Exception:
                log.exception("monitor iteration failed")
            self._stop_event.wait(15)   # check every 15s

    def _reset_at_midnight(self):
        """At 00:00-00:02 IST, anchor day_open_realized to current realized PnL.
        Daily caps then measure delta from this anchor."""
        now = datetime.now()
        if now.hour == 0 and now.minute < 2 and not self.daily_reset_done:
            self.day_open_realized = self._fetch_realized_pnl()
            log.info("Daily reset (00:00 IST): day_open_realized anchored at Rs %.2f",
                     self.day_open_realized)
            self.daily_reset_done = True
        elif now.hour > 0:
            self.daily_reset_done = False

    def _fetch_realized_pnl(self):
        """
        Pair buy/sell trades from tradebook by symbol+product, compute realized PnL.

        Uses FIFO pairing within (symbol, product). Unmatched trades are the
        currently open exposure (already counted by positionbook unrealized).
        """
        try:
            tb = self.client.tradebook()
            rows = tb.get("data", []) if isinstance(tb, dict) else []
        except Exception:
            log.exception("tradebook fetch failed")
            return 0.0

        # Group by (symbol, product), preserve order, separate buys/sells
        from collections import defaultdict, deque
        buys = defaultdict(deque)
        sells = defaultdict(deque)
        for r in rows:
            sym, prod = r.get("symbol"), r.get("product")
            try:
                qty = abs(int(float(r.get("quantity", 0) or 0)))
                price = float(r.get("average_price", 0) or 0)
            except (TypeError, ValueError):
                continue
            if qty == 0 or price == 0:
                continue
            (buys if r.get("action") == "BUY" else sells)[(sym, prod)].append((qty, price))

        realized = 0.0
        for key in set(list(buys.keys()) + list(sells.keys())):
            b = buys[key]
            s = sells[key]
            while b and s:
                bq, bp = b[0]
                sq, sp = s[0]
                matched = min(bq, sq)
                realized += matched * (sp - bp)
                if bq == matched:
                    b.popleft()
                else:
                    b[0] = (bq - matched, bp)
                if sq == matched:
                    s.popleft()
                else:
                    s[0] = (sq - matched, sp)
        return realized

    def _compute_total_pnl(self):
        """Realized (from tradebook) + unrealized (from positionbook)."""
        unrealized = 0.0
        try:
            pb = self.client.positionbook()
            rows = pb.get("data", []) if isinstance(pb, dict) else []
            for r in rows:
                pnl = r.get("pnl")
                if pnl is None:
                    continue
                try:
                    unrealized += float(pnl)
                except (TypeError, ValueError):
                    continue
        except Exception:
            log.exception("positionbook fetch failed")
            unrealized = 0.0
        realized = self._fetch_realized_pnl()
        return realized + unrealized

    def _check_caps(self, pnl):
        cap = self.start_capital
        # Portfolio SL
        psl = self.caps.get("portfolio_sl_pct")
        if psl is not None and pnl <= -abs(psl) * cap:
            return f"PORTFOLIO_SL: pnl={pnl:.2f} <= -{psl*100:.2f}% cap"
        # Portfolio TP
        ptp = self.caps.get("portfolio_tp_pct")
        if ptp is not None and pnl >= abs(ptp) * cap:
            return f"PORTFOLIO_TP: pnl={pnl:.2f} >= +{ptp*100:.2f}% cap"
        # Daily loss / target
        daily_pnl = pnl - self.day_open_realized
        dsl = self.caps.get("daily_loss_pct")
        if dsl is not None and daily_pnl <= -abs(dsl) * cap:
            return f"DAILY_LOSS: daily_pnl={daily_pnl:.2f} <= -{dsl*100:.2f}% cap"
        dtg = self.caps.get("daily_target_pct")
        if dtg is not None and daily_pnl >= abs(dtg) * cap:
            return f"DAILY_TARGET: daily_pnl={daily_pnl:.2f} >= +{dtg*100:.2f}% cap"
        # Max concurrent positions
        mc = self.caps.get("max_concurrent_positions")
        if mc is not None:
            try:
                pb = self.client.positionbook()
                rows = pb.get("data", []) if isinstance(pb, dict) else []
                active = sum(
                    1 for r in rows
                    if int(float(r.get("quantity", 0) or 0)) != 0
                )
                if active > mc:
                    return f"MAX_POSITIONS: active={active} > cap={mc}"
            except Exception:
                log.exception("positionbook fetch for max_positions failed")
        return None

    def join(self):
        try:
            while any(c.is_alive() for c in self.children):
                time.sleep(2)
        except KeyboardInterrupt:
            pass
        self._stop_event.set()
        self.stop_all(reason="user interrupt")