"""ITF executor — DRY_RUN writes to jsonl, live uses Alpaca paper bracket orders. Auto-detect: ALPACA_PAPER_KEY + ALPACA_PAPER_SECRET in env => live mode. Otherwise: dry-run (simulated fill at last quote). Position management: - max 3 open positions per agent - max hold = persona.max_hold_min - EOD flatten at 19:50 UTC (15:50 ET — 10 min before market close) """ from __future__ import annotations import json import os import time import uuid from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Any, Dict, List, Optional, Tuple _HERE = Path(__file__).resolve() try: REPO = _HERE.parents[3] except IndexError: REPO = _HERE.parent # HF Space ORDERS_JSONL = REPO / "data" / "intraday" / "dry-run-orders.jsonl" POSITIONS_PATH = REPO / "data" / "intraday" / "positions.json" BANKROLLS_PATH = REPO / "data" / "intraday" / "agent_bankrolls.json" LEDGER_JSONL = REPO / "data" / "intraday" / "agent_ledger.jsonl" RECON_CURSOR_PATH = REPO / "data" / "intraday" / "fill_reconciliation_cursor.json" POSITIONS_PATH.parent.mkdir(parents=True, exist_ok=True) MAX_OPEN_PER_AGENT = int(os.environ.get("ITF_MAX_OPEN_PER_AGENT", "30")) # 2026-04-25: 5→30 to push 500/day. Was bottleneck — 17 agents × 5 = 85 fleet ceiling, currently 65 → most decisions rejected. New ceiling 17×30 = 510 fleet positions (matches 500/day round-trip target). Env-overridable. EOD_FLATTEN_UTC_HOUR = 19 EOD_FLATTEN_UTC_MIN = 50 # 2026-04-22 — HF Space /app is wiped on every factory_reboot, so the four # attribution files above evaporate. persist_ledgers_to_hub() uploads them # back to the repo at end-of-tick; restore_ledgers.py re-hydrates them on # boot. _LEDGER_DIRTY guards against no-op commits when a tick didn't mutate # anything. The ledger jsonl is append-only: we flip dirty on every append. _LEDGER_DIRTY: bool = False # Max bytes to keep uploading for the append-only ledger before we stop # shipping it every tick (still shipped on the tick that crosses the limit). _LEDGER_MAX_UPLOAD_BYTES = 5 * 1024 * 1024 # 2026-04-24 MARKET-HOURS GATE — 189 broker_errors observed pre-market with # message "options market orders are only allowed during market hours" plus # same-day-expiry asset-not-found. Fix: poll /v2/clock once per 60s and # reject both equity AND options market orders when closed. Re-submission # happens next tick after market opens. _CLOCK_CACHE: tuple[float, bool] | None = None _CLOCK_TTL_SEC = 60.0 def _market_is_open() -> bool: """True iff Alpaca reports the equity market is open. Cached 60s. Fail-closed: any error -> return False (safer than blasting broker_errors). When live_mode() is False (dry run), assume open -- dry run shouldn't depend on external availability.""" if not live_mode(): return True global _CLOCK_CACHE now = time.time() if _CLOCK_CACHE is not None and (now - _CLOCK_CACHE[0] < _CLOCK_TTL_SEC): return _CLOCK_CACHE[1] try: import requests key = os.environ.get("ALPACA_PAPER_KEY", "") secret = os.environ.get("ALPACA_PAPER_SECRET", "") r = requests.get( "https://paper-api.alpaca.markets/v2/clock", headers={"APCA-API-KEY-ID": key, "APCA-API-SECRET-KEY": secret}, timeout=5, ) if r.ok: is_open = bool(r.json().get("is_open")) _CLOCK_CACHE = (now, is_open) return is_open except Exception: pass _CLOCK_CACHE = (now, False) return False # ───── 2026-04-21 v2.5 PER-AGENT SUB-BANKROLL ───── # Each of 14 personas gets an equal slice of current Alpaca equity at cold-start. # Every submit() reserves stake_usd from agent's bankroll; close_expired / # close_position credits stake + realized_pnl back. Enables a SCIENTIFIC # leaderboard (which persona is actually best) instead of a single blended pool. def _load_bankrolls() -> Dict[str, float]: if not BANKROLLS_PATH.exists(): return {} try: return json.loads(BANKROLLS_PATH.read_text()) except Exception: return {} def _save_bankrolls(b: Dict[str, float]) -> None: BANKROLLS_PATH.write_text(json.dumps(b, indent=2, sort_keys=True)) global _LEDGER_DIRTY _LEDGER_DIRTY = True _ALPACA_ACCT_CACHE: Dict[str, Any] = {"ts": 0.0, "data": None} def fetch_alpaca_account(ttl_sec: float = 30.0) -> Dict[str, Any]: """Pull full Alpaca paper account (equity + BP + cash + position MVs) with in-process TTL cache. Returns {} when not in live mode or API fails so callers can decide on fallback. Single source of broker truth — every UI surface (/api/status, /api/bankrolls, /api/leaderboard) and every tick reconcile pulls from here so they never disagree.""" now = time.monotonic() if _ALPACA_ACCT_CACHE["data"] is not None and (now - _ALPACA_ACCT_CACHE["ts"]) < ttl_sec: return _ALPACA_ACCT_CACHE["data"] if not live_mode(): return {} try: import requests r = requests.get( "https://paper-api.alpaca.markets/v2/account", headers={ "APCA-API-KEY-ID": os.environ.get("ALPACA_PAPER_KEY", ""), "APCA-API-SECRET-KEY": os.environ.get("ALPACA_PAPER_SECRET", ""), }, timeout=5, ) if r.ok: j = r.json() data = { "equity": float(j.get("equity") or 0.0), "last_equity": float(j.get("last_equity") or 0.0), "cash": float(j.get("cash") or 0.0), "buying_power": float(j.get("buying_power") or 0.0), "long_market_value": float(j.get("long_market_value") or 0.0), "short_market_value": float(j.get("short_market_value") or 0.0), "daytrade_count": int(j.get("daytrade_count") or 0), "pattern_day_trader": bool(j.get("pattern_day_trader") or False), "fetched_at": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"), } _ALPACA_ACCT_CACHE["data"] = data _ALPACA_ACCT_CACHE["ts"] = now return data except Exception: pass return _ALPACA_ACCT_CACHE.get("data") or {} def _fetch_alpaca_equity() -> float: """Back-compat wrapper. Returns 100_000 fallback when account unavailable.""" a = fetch_alpaca_account() return float(a.get("equity") or 100_000.0) def reconcile_fleet_to_alpaca(min_drift_pct: float = 0.02, persist: bool = True) -> Dict[str, Any]: """Rescale per-agent bankrolls so SUM(agent.bankroll) == alpaca.equity. Drift > min_drift_pct triggers a proportional rescale (each tid scaled by alpaca_equity / internal_total). When persist=True, writes the rescaled ledger back to bankrolls.json so subsequent ticks decide on truth, not drift. When persist=False, returns the would-apply numbers without mutating — used by /api/bankrolls preview path. Returns a stats dict suitable for surfacing in /api/status: {alpaca_equity, internal_before, internal_after, scale, drift_pct, applied (bool), reason (str)} """ acct = fetch_alpaca_account() alpaca_equity = float(acct.get("equity") or 0.0) if alpaca_equity <= 0: return {"applied": False, "reason": "no_alpaca_account", "alpaca_equity": alpaca_equity} b = _load_bankrolls() tids = [k for k in b.keys() if not k.startswith("_")] available = {t: float(b.get(t, 0.0) or 0.0) for t in tids} sum_available = sum(available.values()) # Compute reserved per tid from positions.json (matches /api/bankrolls). positions = _load_positions() or {} reserved_by_tid: Dict[str, float] = {} for tid, plist in positions.items(): for pos in (plist or []): if pos.get("status") == "open": reserved_by_tid[tid] = reserved_by_tid.get(tid, 0.0) + float( pos.get("stake_usd") or 0.0) sum_reserved = sum(reserved_by_tid.values()) internal_total = sum_available + sum_reserved # equity-equivalent if internal_total == 0: return {"applied": False, "reason": "ledger_empty", "alpaca_equity": alpaca_equity, "internal_before": internal_total, "sum_available": round(sum_available, 2), "sum_reserved": round(sum_reserved, 2)} drift = (alpaca_equity - internal_total) / max(1e-6, abs(internal_total)) if abs(drift) < min_drift_pct: return {"applied": False, "reason": "below_threshold", "alpaca_equity": round(alpaca_equity, 2), "internal_before": round(internal_total, 2), "internal_after": round(internal_total, 2), "sum_available": round(sum_available, 2), "sum_reserved": round(sum_reserved, 2), "scale": 1.0, "drift_pct": round(drift, 4)} # Rescale strategy: reserved is anchored to broker positions (we can't move # those without closing). We absorb the drift on the available side so # SUM(available) + SUM(reserved) == alpaca_equity. Each tid's new available # is its current share of total available scaled to the new target. target_available_total = alpaca_equity - sum_reserved if persist: if abs(sum_available) < 1e-6: # Edge case: every penny is reserved. Distribute target equally. share = target_available_total / max(1, len(tids)) for t in tids: b[t] = round(share, 2) else: scale = target_available_total / sum_available for t in tids: b[t] = round(available[t] * scale, 2) meta = b.setdefault("_meta", {}) meta["last_reconcile_at"] = datetime.now(timezone.utc).strftime( "%Y-%m-%dT%H:%M:%SZ") meta["last_reconcile_alpaca_equity"] = round(alpaca_equity, 2) meta["last_reconcile_internal_before"] = round(internal_total, 2) meta["last_reconcile_target_available"] = round(target_available_total, 2) meta["last_reconcile_drift_pct"] = round(drift, 4) _save_bankrolls(b) try: _append_ledger({ "tid": "_FLEET_", "event": "fleet_reconcile", "delta": round(alpaca_equity - internal_total, 2), "balance_before": round(internal_total, 2), "balance_after": round(alpaca_equity, 2), "sum_available_before": round(sum_available, 2), "sum_reserved": round(sum_reserved, 2), "target_available": round(target_available_total, 2), "drift_pct": round(drift, 4), }) except Exception: pass return {"applied": persist, "reason": "drift_above_threshold", "alpaca_equity": round(alpaca_equity, 2), "internal_before": round(internal_total, 2), "internal_after": round(alpaca_equity if persist else internal_total, 2), "sum_available_before": round(sum_available, 2), "sum_reserved": round(sum_reserved, 2), "target_available": round(target_available_total, 2), "drift_pct": round(drift, 4)} def seed_bankrolls(tids: List[str], force: bool = False) -> Dict[str, float]: """Seed each tid at equal share of current Alpaca equity. Idempotent unless `force=True` (used by /api/reset).""" existing = _load_bankrolls() if existing and not force: return existing total = _fetch_alpaca_equity() share = round(total / max(1, len(tids)), 2) b = {tid: share for tid in tids} b["_meta"] = { "seeded_at": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"), "seed_equity_usd": round(total, 2), "seed_share_usd": share, "n_agents": len(tids), } _save_bankrolls(b) return b def get_bankroll(tid: str) -> float: b = _load_bankrolls() return float(b.get(tid, 0.0) or 0.0) def _append_ledger(event: Dict[str, Any]) -> None: """Append a per-agent bankroll event to agent_ledger.jsonl for scientific audit.""" LEDGER_JSONL.parent.mkdir(parents=True, exist_ok=True) event["ts"] = event.get("ts") or datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") with LEDGER_JSONL.open("a") as fh: fh.write(json.dumps(event, default=str) + "\n") global _LEDGER_DIRTY _LEDGER_DIRTY = True def _classify_reject(err: str) -> str: """Map raw Alpaca error body to a short reason_code so the ledger histograms cleanly.""" e = err or "" if "available\":\"0\"" in e or "balance\":\"-" in e: return "insufficient_bp" if "existing_order" in e or "40310000" in e: return "duplicate_order" if "42210000" in e or "qty must be" in e or "qty must" in e: return "qty_invalid" if "base_price" in e: return "limit_too_far" if "wash" in e.lower(): return "wash_trade" if "market_closed" in e or "trading_blocked" in e: return "market_closed" if "422" in e: return "unprocessable_other" if "403" in e: return "forbidden_other" if "429" in e: return "rate_limited" return "other" def _agent_open_reserved(tid: str) -> float: positions = _load_positions() s = 0.0 for pos in (positions.get(tid, []) or []): if pos.get("status") == "open": s += float(pos.get("stake_usd") or 0.0) return s def agent_leverage_check(tid: str, new_stake: float) -> Tuple[bool, str, Dict[str, float]]: """Pre-flight margin/leverage gate for new orders. Returns (ok, reason, stats). Cap formula: `per_agent_cap = (alpaca_equity / n_agents) * leverage_mult`. `n_agents` is the count of seeded tids in bankrolls.json. `leverage_mult` defaults to 4.0 (PDT intraday max) and rises to env `ITF_AGENT_LEVERAGE_MULT` when present. When Alpaca account isn't reachable, falls back to 4× of the seeded share so dry-run + degraded states still gate cleanly. The caller (submit / submit_option) gets a clean rejection ledger event instead of a 40310000 broker error when an agent tries to over-commit. """ try: leverage_mult = float(os.environ.get("ITF_AGENT_LEVERAGE_MULT", "4.0")) except Exception: leverage_mult = 4.0 acct = fetch_alpaca_account() alpaca_equity = float(acct.get("equity") or 0.0) b = _load_bankrolls() tids = [k for k in b.keys() if not k.startswith("_")] n_agents = max(1, len(tids)) if alpaca_equity > 0: per_agent_equity = alpaca_equity / n_agents else: per_agent_equity = float(b.get("_meta", {}).get("seed_share_usd") or 5_500.0) cap = per_agent_equity * leverage_mult reserved = _agent_open_reserved(tid) projected = reserved + float(new_stake or 0.0) stats = { "leverage_mult": round(leverage_mult, 2), "per_agent_equity_usd": round(per_agent_equity, 2), "per_agent_cap_usd": round(cap, 2), "reserved_open_usd": round(reserved, 2), "projected_after_usd": round(projected, 2), "alpaca_equity_usd": round(alpaca_equity, 2), } if projected > cap: return False, (f"agent_leverage_cap projected=${projected:,.0f} > " f"cap=${cap:,.0f} (equity_share=${per_agent_equity:,.0f} × " f"{leverage_mult:.1f}x)"), stats return True, "ok", stats def reserve_bankroll(tid: str, amount: float, meta: Optional[Dict[str, Any]] = None) -> float: """Deduct amount from tid's bankroll. Returns new balance (can go negative — caller checks).""" b = _load_bankrolls() before = float(b.get(tid, 0.0) or 0.0) b[tid] = round(before - amount, 2) _save_bankrolls(b) _append_ledger({"tid": tid, "event": "reserve", "delta": -round(amount, 2), "balance_before": round(before, 2), "balance_after": b[tid], **(meta or {})}) return b[tid] def credit_bankroll(tid: str, amount: float, meta: Optional[Dict[str, Any]] = None) -> float: b = _load_bankrolls() before = float(b.get(tid, 0.0) or 0.0) b[tid] = round(before + amount, 2) _save_bankrolls(b) _append_ledger({"tid": tid, "event": "credit", "delta": round(amount, 2), "balance_before": round(before, 2), "balance_after": b[tid], **(meta or {})}) return b[tid] def all_bankrolls() -> Dict[str, float]: """Return copy without _meta for leaderboard rendering.""" b = _load_bankrolls() return {k: v for k, v in b.items() if not k.startswith("_")} def live_mode() -> bool: """Live only when ITF_MODE=live AND Alpaca keys present. Default is dry_run — safer for an unvalidated key-pair. Explicitly opt in via env `ITF_MODE=live` once you've confirmed the key at /v2/account. """ if os.environ.get("DRY_RUN", "").lower() in ("1", "true", "yes"): return False if os.environ.get("ITF_MODE", "").lower() != "live": return False return bool(os.environ.get("ALPACA_PAPER_KEY") and os.environ.get("ALPACA_PAPER_SECRET")) def _load_positions() -> Dict[str, List[Dict[str, Any]]]: if not POSITIONS_PATH.exists(): return {} try: return json.loads(POSITIONS_PATH.read_text()) except Exception: return {} def _save_positions(p: Dict[str, List[Dict[str, Any]]]) -> None: POSITIONS_PATH.write_text(json.dumps(p, indent=2, default=str)) global _LEDGER_DIRTY _LEDGER_DIRTY = True def _append_order_log(entry: Dict[str, Any]) -> None: ORDERS_JSONL.parent.mkdir(parents=True, exist_ok=True) with ORDERS_JSONL.open("a") as fh: fh.write(json.dumps(entry, default=str) + "\n") def refresh_broker_statuses() -> Dict[str, int]: """Re-poll Alpaca for every position with a broker_order_id whose cached broker_status is non-terminal. Updates positions.json in place. Terminal statuses (skipped to save API calls): filled, canceled, expired, rejected, replaced, closed_by_agent, sim_*, closed. Returns a counter dict of what changed, e.g. {"polled": 23, "updated": 18, "filled": 11, "canceled": 2}. Called once at the top of tick_once() so the /api/status view never shows stale pending_new. """ stats = {"polled": 0, "updated": 0, "filled": 0, "canceled": 0, "other": 0, "errors": 0, "budget_exceeded": 0} if not live_mode(): return stats key = os.environ.get("ALPACA_PAPER_KEY") secret = os.environ.get("ALPACA_PAPER_SECRET") if not (key and secret): return stats TERMINAL = { "filled", "canceled", "cancelled", "expired", "rejected", "replaced", "closed_by_agent", "closed", "done_for_day", "stopped", "suspended", "not_found", } import requests headers = {"APCA-API-KEY-ID": key, "APCA-API-SECRET-KEY": secret} # 2026-04-22 — time-budget guard. Without this, a flaky Alpaca paper-api # with N non-terminal orders × 6s per-call could block tick_once() for # ~N*6s (observed: ~16 min stall at tick 1 after 232 daytrades piled up). budget_sec = float(os.environ.get("ITF_REFRESH_BROKER_BUDGET_SEC", "20")) deadline = time.monotonic() + budget_sec positions = _load_positions() changed = False for agent_tid, lst in positions.items(): for p in lst: if time.monotonic() >= deadline: stats["budget_exceeded"] += 1 if changed: _save_positions(positions) return stats oid = p.get("broker_order_id") if not oid: continue cur = (p.get("broker_status") or "").lower() if cur in TERMINAL: continue stats["polled"] += 1 try: r = requests.get( f"https://paper-api.alpaca.markets/v2/orders/{oid}", headers=headers, timeout=6, ) if r.status_code == 404: p["broker_status"] = "not_found" stats["updated"] += 1 stats["other"] += 1 changed = True continue if not r.ok: stats["errors"] += 1 continue body = r.json() new = (body.get("status") or "").lower() if new and new != cur: p["broker_status"] = new if body.get("filled_avg_price"): p["filled_avg_price"] = float(body["filled_avg_price"]) if body.get("filled_at"): p["filled_at"] = body["filled_at"] if body.get("filled_qty"): p["filled_qty"] = float(body["filled_qty"]) stats["updated"] += 1 if new == "filled": stats["filled"] += 1 elif new in {"canceled", "cancelled"}: stats["canceled"] += 1 else: stats["other"] += 1 changed = True except Exception: stats["errors"] += 1 if changed: _save_positions(positions) return stats def close_stale_losers(max_age_sec: int = 14400, min_loss_pct: float = 0.02) -> Dict[str, Any]: """Close equity positions older than max_age_sec with unrealized PnL <= -min_loss_pct. 2026-04-22 ROUND-2 BP UNLOCK — Alpaca paper `insufficient balance` + free_bp=$0 while total cash sits at $49K = open positions consumed all free BP. Nothing in the fleet closes stale losers proactively (MIN_HOLD_SEC=900 only prevents churn; EOD-flatten doesn't fire intra-day). This helper sweeps equity positions that are both >max_age_sec old AND underwater >=min_loss_pct, so the 70% deploy target can actually reserve BP. Safety: * Respects MIN_HOLD_SEC implicitly (default 4h >> 15min) * Equities only — crypto (BTC/USD etc) use non-USD margin, no BP pressure * Time-budgeted at ITF_CLOSE_STALE_BUDGET_SEC (default 10s) so a flaky Alpaca API can't stall tick_once() * Credits reserved stake back to agent sub-bankroll via credit_bankroll() (conservative — realized PnL reconciles via reconcile_broker_fills next tick) Returns: {closed, pnl_freed_usd, errors, skipped_too_young, skipped_winning, skipped_crypto, budget_exceeded}. """ budget_sec = float(os.environ.get("ITF_CLOSE_STALE_BUDGET_SEC", "10")) deadline = time.monotonic() + budget_sec stats = { "closed": 0, "pnl_freed_usd": 0.0, "errors": 0, "skipped_too_young": 0, "skipped_winning": 0, "skipped_crypto": 0, "budget_exceeded": 0, } if not live_mode(): return stats key = os.environ.get("ALPACA_PAPER_KEY") secret = os.environ.get("ALPACA_PAPER_SECRET") if not (key and secret): return stats import requests headers = {"APCA-API-KEY-ID": key, "APCA-API-SECRET-KEY": secret} now_utc = datetime.now(timezone.utc) positions = _load_positions() dirty = False try: for agent_tid, lst in list(positions.items()): for p in list(lst): if time.monotonic() >= deadline: stats["budget_exceeded"] += 1 if dirty: _save_positions(positions) return stats try: if p.get("status") != "open": continue ticker = (p.get("ticker") or "").strip() if not ticker: continue # Skip crypto — settles in non-USD margin, no BP pressure. if _asset_class(ticker) == "crypto": stats["skipped_crypto"] += 1 continue # Age check. Accept either `opened_at` or `ts`; both ISO-Z. opened_raw = p.get("opened_at") or p.get("ts") or "" try: opened_dt = datetime.fromisoformat( str(opened_raw).replace("Z", "+00:00") ) except Exception: # unreadable timestamp → don't close (treat as young) stats["skipped_too_young"] += 1 continue age_sec = (now_utc - opened_dt).total_seconds() if age_sec < max_age_sec: stats["skipped_too_young"] += 1 continue # Fetch live position to get unrealized_plpc. If position not # found at broker, skip — fill reconciler will tidy it up. try: r = requests.get( f"https://paper-api.alpaca.markets/v2/positions/{ticker}", headers=headers, timeout=6, ) except Exception: stats["errors"] += 1 continue if r.status_code == 404: # Broker doesn't have it — nothing to close, our ledger # will self-heal via the reconciler. continue if not r.ok: stats["errors"] += 1 continue try: live = r.json() upl = float(live.get("unrealized_plpc") or 0) mv = float(live.get("market_value") or 0) except Exception: stats["errors"] += 1 continue if upl > -min_loss_pct: stats["skipped_winning"] += 1 continue # Close via positions endpoint (net-flat, correct for bracket). try: rd = requests.delete( f"https://paper-api.alpaca.markets/v2/positions/{ticker}", headers=headers, timeout=10, ) if not rd.ok: stats["errors"] += 1 continue except Exception: stats["errors"] += 1 continue # Mark local position closed + credit reserved stake back. p["status"] = "closed_stale_loser" p["closed_at"] = now_utc.strftime("%Y-%m-%dT%H:%M:%SZ") p["close_reason"] = f"stale_loser age={int(age_sec)}s upl={upl:.4f}" dirty = True stake_portion = float(p.get("stake_usd") or 0) if stake_portion > 0: try: credit_bankroll(agent_tid, stake_portion, meta={ "ticker": ticker, "event_type": "stale_loser_close", "age_sec": int(age_sec), "unrealized_plpc": round(upl, 4), "market_value_usd": round(mv, 2), }) except Exception: pass # Also log to order log for audit parity with close_position(). try: _append_order_log({ "ts": p["closed_at"], "agent_tid": agent_tid, "ticker": ticker, "action": "close_stale_loser", "age_sec": int(age_sec), "unrealized_plpc": round(upl, 4), "market_value_usd": round(mv, 2), "mode": "live", }) except Exception: pass stats["closed"] += 1 stats["pnl_freed_usd"] += mv except Exception: stats["errors"] += 1 finally: if dirty: _save_positions(positions) return stats # ───── 2026-04-22 ROUND-3 ORDER-PILEUP GUARDS ───── # Incident: 319 open bracket orders stacked on Alpaca paper (SPY×61, QQQ×37, # NVDA×35…) consumed $63K of initial_margin. daytrading_buying_power fell to # $246 on $101K equity. Agents kept emitting bracket orders every tick; none # filled (limit prices drifted, or same-symbol contention). Manual # `DELETE /v2/orders` freed BP back to $157K. Without these three guards the # pile rebuilds within hours. # # Guard 1: _refresh_pending_count() — 30s cached map of open-orders-by-symbol # Guard 2: cancel_stale_pending() — cancels orders older than max_age_min # Guard 3: _bp_pre_check() — rejects new placements when free BP < $500 # # All three fail-open (errors logged, never raise). _PENDING_BY_SYMBOL: Dict[str, int] = {} _PENDING_BY_SYMBOL_TS: float = 0.0 _PENDING_CACHE_TTL_SEC: float = 30.0 def _refresh_pending_count(force: bool = False) -> None: """Refresh the open-order-by-symbol cache. 30s TTL to keep HTTP traffic bounded while still catching pileup within a single tick.""" global _PENDING_BY_SYMBOL, _PENDING_BY_SYMBOL_TS if not live_mode(): return if not force and (time.time() - _PENDING_BY_SYMBOL_TS) < _PENDING_CACHE_TTL_SEC: return key = os.environ.get("ALPACA_PAPER_KEY") secret = os.environ.get("ALPACA_PAPER_SECRET") if not (key and secret): return try: import requests r = requests.get( "https://paper-api.alpaca.markets/v2/orders", headers={"APCA-API-KEY-ID": key, "APCA-API-SECRET-KEY": secret}, params={"status": "open", "limit": 500}, timeout=6, ) if not r.ok: return orders = r.json() or [] counts: Dict[str, int] = {} for o in orders: sym = (o.get("symbol") or "").strip() if not sym: continue counts[sym] = counts.get(sym, 0) + 1 _PENDING_BY_SYMBOL = counts _PENDING_BY_SYMBOL_TS = time.time() except Exception: # Fail-open: stale cache is fine, will refresh next tick. pass def _pending_count_for(symbol: str) -> int: """Return cached pending-order count for a symbol. Refreshes if stale.""" _refresh_pending_count() return int(_PENDING_BY_SYMBOL.get(symbol, 0)) def _get_daytrading_buying_power() -> float: """Fetch Alpaca daytrading_buying_power. Returns 0.0 on error (fail-closed for BP-pre-check: if we can't read BP, treat as starved so we don't pile up).""" if not live_mode(): return 1_000_000.0 # dry-run: unlimited key = os.environ.get("ALPACA_PAPER_KEY") secret = os.environ.get("ALPACA_PAPER_SECRET") if not (key and secret): return 0.0 try: import requests r = requests.get( "https://paper-api.alpaca.markets/v2/account", headers={"APCA-API-KEY-ID": key, "APCA-API-SECRET-KEY": secret}, timeout=5, ) if not r.ok: return 0.0 j = r.json() # 2026-04-25: Alpaca returns BP fields as STRINGS ('0' is truthy in # Python!), so `daytrading_buying_power or buying_power` returned '0' # whenever PDT was exhausted. Cast each field to float FIRST, then # pick max(daytrading_bp, regt_bp) so we use whichever is usable. # daytrading_bp is intraday-only quota (often 0 when PDT limit hit), # buying_power = regt_buying_power = settled-margin BP that survives # PDT exhaustion. Equity > $25K so we are NOT subject to the PDT # 4-roundtrip rule; daytrade_count of 69 is just paper-tracking noise. try: dt_bp = float(j.get("daytrading_buying_power") or 0) except Exception: dt_bp = 0.0 try: bp = float(j.get("buying_power") or 0) except Exception: bp = 0.0 return max(dt_bp, bp) except Exception: return 0.0 def _bp_pre_check(min_bp: float = 500.0) -> Dict[str, Any]: """Return {ok: bool, bp: float, reason: str}. If BP is below `min_bp` the caller should skip placement this tick. If BP is below 5% of equity, we also trigger a fast cancel_stale_pending(max_age_min=10) to unblock.""" out = {"ok": True, "bp": 0.0, "reason": ""} if not live_mode(): out["bp"] = 1_000_000.0 return out bp = _get_daytrading_buying_power() out["bp"] = bp equity = _fetch_alpaca_equity() if bp < min_bp: out["ok"] = False out["reason"] = f"bp={bp:.2f} < min={min_bp:.2f}" if equity > 0 and bp < 0.05 * equity: # BP less than 5% of equity — the pileup is eating margin. Force an # inline 10-min stale cancel to unblock, best-effort. try: cs = cancel_stale_pending(max_age_min=10) out["inline_stale_cancel"] = cs except Exception as e: out["inline_stale_cancel_err"] = str(e)[:200] return out def cancel_stale_pending(max_age_min: int = 30) -> Dict[str, Any]: """Cancel Alpaca open orders older than `max_age_min` minutes. 2026-04-22 ROUND-3 — Pileup RCA: Alpaca paper accumulated 319 open brackets on ~12 symbols (SPY×61, QQQ×37, etc). None filled — limit prices drifted or same-symbol contention blocked sequencing. initial_margin ate BP down to $246 on $101K equity. This helper sweeps every tick/10 so the pile never rebuilds. Safety: * Live-mode only (no dry-run no-op noise) * Time-budgeted at ITF_CANCEL_STALE_BUDGET_SEC (default 10s) * Only cancels orders where `submitted_at > max_age_min min ago` * Uses DELETE /v2/orders/{id}; logs non-OK per-order but continues * DOES cancel during pre-open (caller controls age threshold; legit GTC overnight orders aren't stale at 30min so default is safe) Returns: {cancelled, errors, skipped_young, budget_exceeded, seen}. """ stats = {"cancelled": 0, "errors": 0, "skipped_young": 0, "budget_exceeded": 0, "seen": 0} if not live_mode(): return stats key = os.environ.get("ALPACA_PAPER_KEY") secret = os.environ.get("ALPACA_PAPER_SECRET") if not (key and secret): return stats budget_sec = float(os.environ.get("ITF_CANCEL_STALE_BUDGET_SEC", "10")) deadline = time.monotonic() + budget_sec import requests headers = {"APCA-API-KEY-ID": key, "APCA-API-SECRET-KEY": secret} now_utc = datetime.now(timezone.utc) try: r = requests.get( "https://paper-api.alpaca.markets/v2/orders", headers=headers, params={"status": "open", "limit": 500}, timeout=8, ) if not r.ok: stats["errors"] += 1 return stats orders = r.json() or [] except Exception: stats["errors"] += 1 return stats stats["seen"] = len(orders) for o in orders: if time.monotonic() >= deadline: stats["budget_exceeded"] += 1 break try: oid = o.get("id") subm = o.get("submitted_at") or o.get("created_at") or "" if not (oid and subm): continue try: subm_dt = datetime.fromisoformat(str(subm).replace("Z", "+00:00")) except Exception: continue age_min = (now_utc - subm_dt).total_seconds() / 60.0 if age_min < max_age_min: stats["skipped_young"] += 1 continue try: rd = requests.delete( f"https://paper-api.alpaca.markets/v2/orders/{oid}", headers=headers, timeout=6, ) if rd.status_code in (200, 204, 207): stats["cancelled"] += 1 else: stats["errors"] += 1 except Exception: stats["errors"] += 1 except Exception: stats["errors"] += 1 # Force-refresh the per-symbol cache so the next placement sees post-cancel state. try: _refresh_pending_count(force=True) except Exception: pass return stats def _asset_class(ticker: str) -> str: if "/" in ticker: return "crypto" return "equity" def _make_client_order_id(agent_tid: str, ticker: str) -> str: """2026-04-22 — every outbound order gets a deterministic client_order_id so the reconciler can trace a broker fill back to the right agent even if the local positions.json is wiped. Format: "::". Alpaca caps this at 128 chars; normalize ticker (no "/" etc) to be safe.""" safe_ticker = (ticker or "").replace("/", "-").replace(" ", "")[:16].upper() safe_tid = (agent_tid or "anon").replace(":", "-")[:32] return f"{safe_tid}:{safe_ticker}:{uuid.uuid4().hex[:8]}" def _alpaca_place_bracket(ticker: str, qty: float, stake: float, last: float, side: str, stop_price: float, tp_price: float, client_order_id: Optional[str] = None) -> Dict[str, Any]: """Place an Alpaca paper order. Routing (canonical alpaca-py examples pattern): * crypto (BTC/USD etc) → market GTC, fractional qty ok, NO bracket * equity qty >= 1 integer → bracket with integer qty + stop_loss + take_profit * equity qty < 1 or frac → notional-based market day, NO bracket (Alpaca 422s on bracket+fractional; stop/TP tracked client-side in close_expired) """ import requests key = os.environ["ALPACA_PAPER_KEY"] secret = os.environ["ALPACA_PAPER_SECRET"] asset = _asset_class(ticker) headers = {"APCA-API-KEY-ID": key, "APCA-API-SECRET-KEY": secret} if asset == "crypto": payload = { "symbol": ticker, "qty": qty, "side": side, "type": "market", "time_in_force": "gtc", } else: int_qty = int(qty) # floor — Alpaca rejects bracket on fractional is_short = (side == "sell") # 2026-04-25 PQTF-LESSON: bracket equity orders require DTBP, not regt # BP. When PDT-DTBP is starved (=0 from broker even though regt BP is # plentiful), Alpaca rejects 40310000. PQTF compounded $244K because it # held option positions OVERNIGHT (no day-trade classification, no DTBP # cap). For ITF equities, ITF_PREFER_NON_BRACKET=1 routes to simple # market+integer-qty (regt BP path) instead of bracket-day. Stops are # tracked client-side by close_stale_losers + close_expired. prefer_non_bracket = (os.environ.get("ITF_PREFER_NON_BRACKET", "0") or "0") not in ("0", "", "false", "False") # Auto-fallback: if DTBP < estimated cost for this trade, use non-bracket # path even when ITF_PREFER_NON_BRACKET is unset. dt_bp_starved = False if int_qty >= 1 and live_mode(): try: _r = requests.get( "https://paper-api.alpaca.markets/v2/account", headers=headers, timeout=5, ) if _r.ok: _j = _r.json() try: _dt = float(_j.get("daytrading_buying_power") or 0) except Exception: _dt = 0.0 if _dt < stake: dt_bp_starved = True except Exception: pass use_non_bracket = prefer_non_bracket or dt_bp_starved if int_qty >= 1 and not use_non_bracket: # Bracket day-trade path. DTBP must cover stake. payload = { "symbol": ticker, "qty": int_qty, "side": side, "type": "market", "time_in_force": "day", "order_class": "bracket", "extended_hours": False, "stop_loss": {"stop_price": round(stop_price, 2)}, "take_profit": {"limit_price": round(tp_price, 2)}, } elif int_qty >= 1 and use_non_bracket: # Simple integer market — uses regt BP, integer qty (works for short # too: Alpaca disallows fractional shorts but accepts integer-qty # shorts with regular BP). payload = { "symbol": ticker, "qty": int_qty, "side": side, "type": "market", "time_in_force": "day", } else: # int_qty < 1 — only longs can use notional fractional. Shorts must # be qty>=1 (Alpaca: "fractional orders cannot be sold short"). # # 2026-04-28 MAX-AGGRO upsize — instead of dropping the trade, when # ITF_MAX_AGGRO=1 and side=short with int_qty=0, upsize qty to 1. # Cost: stake rises from agent's intended size to 1×last_quote; # since user authorised full-loss tolerance, executing the short # is preferable to discarding the directional view (46+ skips # observed in last 300 ledger events 2026-04-28). if is_short: if os.environ.get("ITF_MAX_AGGRO", "0") == "1": int_qty = 1 payload = { "symbol": ticker, "qty": int_qty, "side": side, "type": "market", "time_in_force": "day", } else: raise ValueError( f"short fractional skip: ticker={ticker} qty={qty} stake={stake} " f"(int_qty={int_qty} < 1, Alpaca rejects fractional shorts)" ) else: payload = { "symbol": ticker, "notional": round(stake, 2), "side": side, "type": "market", "time_in_force": "day", } if client_order_id: payload["client_order_id"] = client_order_id r = requests.post( "https://paper-api.alpaca.markets/v2/orders", headers=headers, json=payload, timeout=10, ) r.raise_for_status() return r.json() def submit(agent_tid: str, order: Dict[str, Any], last_quote: float) -> Dict[str, Any]: """Submit an order. Shape of `order`: {ticker, side: long|short, stake_usd, stop_pct, take_profit_pct, thesis} Returns the recorded order entry (with fill or simulated fill). """ # 2026-04-28 v8 — REMOVED v7 DTBP pre-flight refuse. Root cause was not # DTBP exhaustion; it was Alpaca account config `dtbp_check: 'entry'` # forcing every equity entry to compare cost_basis against DTBP=$400. # User flipped dtbp_check + pdt_check to 'exit' + ptp_no_exception_entry # to false → equity entries now use RegT BP ($36K) not DTBP. Old v7 refuse # would block entries the broker would now accept. positions = _load_positions() open_for_agent = [p for p in positions.get(agent_tid, []) if p.get("status") == "open"] # 2026-04-21 wash-trade pre-check: if same agent has an OPEN opposite-side # position on same ticker, skip. Alpaca rejects these as wash-trades anyway; # pre-check gives a clean ledger event instead of a broker_error. _inbound_side = order.get("side") _inbound_ticker = order.get("ticker") for _p in open_for_agent: if _p.get("ticker") == _inbound_ticker and _p.get("side") and _inbound_side and _p.get("side") != _inbound_side: wash = { "ts": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"), "agent_tid": agent_tid, "status": "wash_skip", "reason": f"opposite-side {_p.get('side')} open on {_inbound_ticker}", "order": order, } _append_order_log(wash) return wash # 2026-04-25 cross-agent same-ticker collision pre-check. After non-bracket # fix unblocked DTBP, the next failure layer was: agent A long QQQ, agent B # tries short QQQ → Alpaca account-wide position pool rejects with # "insufficient qty available" (40310000 with `available`/`existing_qty` # fields). Pre-check returns clean ledger event so agent retries elsewhere # next tick (or divert pool kicks in upstream). Cross-agent SAME-side is # allowed (multiple longs aggregate). Only OPPOSITE side is blocked. if _inbound_side and _inbound_ticker: for _other_tid, _other_rows in positions.items(): if _other_tid == agent_tid: continue for _op in _other_rows: if _op.get("status") != "open": continue if _op.get("ticker") != _inbound_ticker: continue _other_side = _op.get("side") if _other_side and _other_side != _inbound_side: skip = { "ts": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"), "agent_tid": agent_tid, "status": "cross_agent_collision_skip", "reason": f"agent {_other_tid} has opposite-side {_other_side} open on {_inbound_ticker}", "order": order, } _append_order_log(skip) return skip # break inner loop after first match for this other_tid (no need to # walk all their positions) # 2026-04-25 fleet diversification cap — limit how many agents can take the # SAME (ticker, side) so the fleet spreads across the universe instead of # piling 8 agents long QQQ. Skipping forces the agent to pass this tick; # the upstream divert pool then routes them to a different ticker. Tunable # via ITF_MAX_AGENTS_PER_DIR_TICKER (default 3 agents per direction). _max_agents_dir = int(os.environ.get("ITF_MAX_AGENTS_PER_DIR_TICKER", "3")) if _inbound_side and _inbound_ticker and _max_agents_dir > 0: _same_dir_count = 0 _same_dir_tids = [] for _other_tid, _other_rows in positions.items(): if _other_tid == agent_tid: continue for _op in _other_rows: if (_op.get("status") == "open" and _op.get("ticker") == _inbound_ticker and _op.get("side") == _inbound_side): _same_dir_count += 1 _same_dir_tids.append(_other_tid) break # one match per other agent is enough if _same_dir_count >= _max_agents_dir: skip = { "ts": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"), "agent_tid": agent_tid, "status": "fleet_dir_cap_skip", "reason": (f"{_same_dir_count} agents already {_inbound_side} on " f"{_inbound_ticker} >= cap {_max_agents_dir} " f"({','.join(_same_dir_tids[:3])})"), "order": order, } _append_order_log(skip) return skip if len(open_for_agent) >= MAX_OPEN_PER_AGENT: reject = { "ts": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"), "agent_tid": agent_tid, "status": "rejected", "reason": f"max {MAX_OPEN_PER_AGENT} open positions already", "order": order, } _append_order_log(reject) return reject # 2026-04-22 ROUND-3 GUARD 1 — per-symbol dedup. Before we even build the # order payload, check how many OPEN orders already exist at Alpaca for # this symbol. If >= ITF_MAX_PENDING_PER_SYMBOL (default 2), skip: another # bracket will just deepen the 319-order pileup. # 2026-04-28 — leverage gate. Refuse if agent's projected reserved would # exceed (per_agent_equity × ITF_AGENT_LEVERAGE_MULT). Default 4× = PDT max. _lvg_ok, _lvg_reason, _lvg_stats = agent_leverage_check( agent_tid, float(order.get("stake_usd") or 0.0)) if not _lvg_ok: skip = { "ts": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"), "agent_tid": agent_tid, "status": "leverage_cap_skip", "reason": _lvg_reason, "leverage_stats": _lvg_stats, "order": order, } _append_order_log(skip) return skip max_pending = int(os.environ.get("ITF_MAX_PENDING_PER_SYMBOL", "2")) if live_mode(): sym_for_check = str(order.get("ticker") or "") try: pend = _pending_count_for(sym_for_check) except Exception: pend = 0 if pend >= max_pending: skip = { "ts": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"), "agent_tid": agent_tid, "status": "broker_skip_dedupe", "reason": f"already {pend} open orders for {sym_for_check} " f">= max_pending_per_symbol={max_pending}", "order": order, } _append_order_log(skip) return skip # 2026-04-22 ROUND-3 GUARD 3 — BP pre-check. If daytrading_buying_power # is starved (<$500), refuse and let cancel_stale_pending reclaim BP # before we try again next tick. try: bp_ok = _bp_pre_check(min_bp=float(os.environ.get("ITF_MIN_BP_USD", "500"))) except Exception: bp_ok = {"ok": True, "bp": -1.0, "reason": ""} if not bp_ok.get("ok"): skip = { "ts": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"), "agent_tid": agent_tid, "status": "broker_skip_bp_starved", "reason": bp_ok.get("reason", "bp low"), "bp": round(float(bp_ok.get("bp") or 0), 2), "inline_stale_cancel": bp_ok.get("inline_stale_cancel"), "order": order, } _append_order_log(skip) return skip ticker = order["ticker"] side = order["side"] # "long" | "short" stake = float(order.get("stake_usd", 1000)) stop_pct = float(order.get("stop_pct", 0.005)) # 2026-04-26 — sub-bankroll cap. Reject if reserved_open + new_stake > sub-bankroll. # Was bleeding because momentum-1/mean-rev-1/earnings-gap-1 went NEGATIVE bk # by stacking 6+ open positions on a $5,811 sub-pot (audit shows mean-rev-1 # at -$1,350 with 6 open positions × ~$5K each = $30K reserved). Enforce a # hard ceiling: sum(open stakes) + new_stake ≤ get_bankroll(tid). try: _agent_avail = get_bankroll(agent_tid) _reserved = 0.0 for _p in open_for_agent: try: _reserved += float(_p.get("stake_usd") or 0) except Exception: continue _total_committed = _reserved + stake # 2026-04-26 PM — user wants leverage. Cap raised 1.05× → 3.0× sub-bankroll. # Real Alpaca paper account has ~4× margin available; 3× per-agent is # conservative-aggressive. Fleet still capped at MAX_OPEN_PER_AGENT. _leverage_mult = float(os.environ.get('ITF_AGENT_LEVERAGE', '3.0')) _cap = (_agent_avail + _reserved) * _leverage_mult if _total_committed > _cap: skip = { "ts": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"), "agent_tid": agent_tid, "status": "sub_bankroll_cap_skip", "reason": f"reserved=${_reserved:.0f} + new=${stake:.0f} " f"would exceed sub-bankroll ${_agent_avail+_reserved:.0f} (cap ${_cap:.0f})", "reserved_open": round(_reserved, 2), "agent_available": round(_agent_avail, 2), "order": order, } _append_order_log(skip) return skip except Exception: pass # fail-open if bookkeeping unavailable tp_pct = float(order.get("take_profit_pct", 0.012)) last = float(last_quote or 0) or 1.0 qty = round(stake / last, 2) if side == "long": stop_price = last * (1 - stop_pct) tp_price = last * (1 + tp_pct) alp_side = "buy" else: stop_price = last * (1 + stop_pct) tp_price = last * (1 - tp_pct) alp_side = "sell" client_order_id = _make_client_order_id(agent_tid, ticker) entry = { "ts": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"), "agent_tid": agent_tid, "ticker": ticker, "side": side, "qty": qty, "entry_price": round(last, 4), "stop_price": round(stop_price, 4), "take_profit_price": round(tp_price, 4), "stake_usd": round(stake, 2), "stop_pct": stop_pct, "take_profit_pct": tp_pct, "thesis": order.get("thesis", "")[:500], "status": "open", "mode": "live" if live_mode() else "dry_run", "client_order_id": client_order_id, } if live_mode(): try: resp = _alpaca_place_bracket(ticker, qty, stake, last, alp_side, stop_price, tp_price, client_order_id=client_order_id) entry["broker_order_id"] = resp.get("id") entry["broker_status"] = resp.get("status") entry["broker_class"] = resp.get("order_class") or ("notional" if resp.get("notional") else "bracket") # 2026-04-22 ROUND-3 — optimistically bump the per-symbol pending # cache so a second agent in the same tick will see the dedup guard. try: _PENDING_BY_SYMBOL[ticker] = int(_PENDING_BY_SYMBOL.get(ticker, 0)) + 1 except Exception: pass except Exception as e: entry["status"] = "broker_error" # Capture Alpaca body text when available (RequestException.response) so we can # diagnose broker_errors beyond the generic "422 Unprocessable Entity" prefix. body = "" try: body = " | body=" + getattr(e, "response", None).text[:400] # type: ignore except Exception: pass entry["error"] = (str(e) + body)[:600] # 2026-04-25 — mirror reject into agent_ledger so per-agent visibility # exists without correlating against positions.json (5MB+ blob). _append_ledger({ "tid": agent_tid, "event": "broker_reject", "ticker": ticker, "side": entry.get("side"), "stake": entry.get("stake_usd"), "instrument": "equity_or_crypto", "reason_code": _classify_reject(entry["error"]), "reason": entry["error"][:300], }) else: # Dry run — simulate the fill and set sim_close_at for EOD flatten entry["sim_filled_at"] = last entry["sim_pnl_usd"] = 0.0 # filled flat, realized on close positions.setdefault(agent_tid, []).append(entry) _save_positions(positions) _append_order_log(entry) # v2.5 — reserve stake from agent's sub-bankroll (do NOT reserve if broker rejected). if entry.get("status") == "open": new_bal = reserve_bankroll(agent_tid, stake, meta={ "ticker": ticker, "side": side, "stake": round(stake, 2), "instrument": "equity_or_crypto", "broker_order_id": entry.get("broker_order_id"), }) entry["agent_bankroll_after_reserve"] = new_bal return entry def _occ_symbol(underlying: str, expiry: datetime, option_type: str, strike: float) -> str: """OCC-standard option symbol: . Example: SPY251220C00480000 = SPY call, strike $480, expiring 2025-12-20. """ exp = expiry.strftime("%y%m%d") cp = "C" if option_type.lower().startswith("c") else "P" strike_int = int(round(strike * 1000)) return f"{underlying.upper()}{exp}{cp}{strike_int:08d}" def _next_expiry(dte: int, now_utc: datetime) -> datetime: """Return the nearest US market expiry that is `dte` trading days ahead. SPY/QQQ/IWM have daily expiries (0/1/2 DTE) during the week; we approximate by skipping weekends only (holidays treated as weekdays for dry-run intent).""" d = now_utc added = 0 while added < max(0, dte): d = d + timedelta(days=1) if d.weekday() < 5: added += 1 return d def submit_option(agent_tid: str, order: Dict[str, Any], last_quote: float) -> Dict[str, Any]: """Submit an intraday option order. Shape: {underlying, option_type, strategy, dte, strike_offset_pct, wing_width_pct, stake_usd, max_loss_pct, thesis} Strategy handling: - "long": 1-leg long call/put - "vertical_debit": 2-leg debit spread (buy ATM, sell ATM+wing) - "vertical_credit": 2-leg credit spread (sell ATM, buy ATM+wing) - "iron_condor": 4-leg (call spread above, put spread below) - "straddle": 2-leg long call + long put at same strike Dry-run logs the structured intent with computed OCC symbols. Live mode routes to Alpaca /v2/options/orders (minimal wrapper; paper-only). """ positions = _load_positions() open_for_agent = [p for p in positions.get(agent_tid, []) if p.get("status") == "open"] # 2026-04-28 — same leverage gate as equity submit() (mirror). _lvg_ok, _lvg_reason, _lvg_stats = agent_leverage_check( agent_tid, float(order.get("stake_usd") or 0.0)) if not _lvg_ok: skip = { "ts": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"), "agent_tid": agent_tid, "status": "leverage_cap_skip", "reason": _lvg_reason, "leverage_stats": _lvg_stats, "order": order, } _append_order_log(skip) return skip if len(open_for_agent) >= MAX_OPEN_PER_AGENT: reject = { "ts": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"), "agent_tid": agent_tid, "status": "rejected", "reason": f"max {MAX_OPEN_PER_AGENT} open positions already", "order": order, } _append_order_log(reject) return reject now = datetime.now(timezone.utc) underlying = order["underlying"] option_type = order.get("option_type", "call") strategy = order.get("strategy", "long") dte = int(order.get("dte", 0) or 0) offset_pct = float(order.get("strike_offset_pct", 0.0) or 0.0) wing_pct = float(order.get("wing_width_pct", 0.01) or 0.01) stake = float(order.get("stake_usd", 500)) last = float(last_quote or 0) or 1.0 expiry = _next_expiry(dte, now) # Compute strikes (rounded to $1 — broker will snap to chain) atm = round(last * (1 + offset_pct)) wing_up = round(last * (1 + offset_pct + wing_pct)) wing_dn = round(last * (1 + offset_pct - wing_pct)) legs: List[Dict[str, Any]] = [] if strategy == "long": legs = [{"side": "buy", "symbol": _occ_symbol(underlying, expiry, option_type, atm), "qty": 1}] elif strategy == "vertical_debit": outer = wing_up if option_type == "call" else wing_dn legs = [ {"side": "buy", "symbol": _occ_symbol(underlying, expiry, option_type, atm), "qty": 1}, {"side": "sell", "symbol": _occ_symbol(underlying, expiry, option_type, outer), "qty": 1}, ] elif strategy == "vertical_credit": outer = wing_up if option_type == "call" else wing_dn legs = [ {"side": "sell", "symbol": _occ_symbol(underlying, expiry, option_type, atm), "qty": 1}, {"side": "buy", "symbol": _occ_symbol(underlying, expiry, option_type, outer), "qty": 1}, ] elif strategy == "iron_condor": legs = [ # call spread above {"side": "sell", "symbol": _occ_symbol(underlying, expiry, "call", wing_up), "qty": 1}, {"side": "buy", "symbol": _occ_symbol(underlying, expiry, "call", round(last*(1+offset_pct+2*wing_pct))), "qty": 1}, # put spread below {"side": "sell", "symbol": _occ_symbol(underlying, expiry, "put", wing_dn), "qty": 1}, {"side": "buy", "symbol": _occ_symbol(underlying, expiry, "put", round(last*(1+offset_pct-2*wing_pct))), "qty": 1}, ] elif strategy == "straddle": legs = [ {"side": "buy", "symbol": _occ_symbol(underlying, expiry, "call", atm), "qty": 1}, {"side": "buy", "symbol": _occ_symbol(underlying, expiry, "put", atm), "qty": 1}, ] else: legs = [{"side": "buy", "symbol": _occ_symbol(underlying, expiry, option_type, atm), "qty": 1}] entry = { "ts": now.strftime("%Y-%m-%dT%H:%M:%SZ"), "agent_tid": agent_tid, "instrument_type": "option", "underlying": underlying, "option_type": option_type, "strategy": strategy, "dte": dte, "expiry": expiry.strftime("%Y-%m-%d"), "atm_strike": atm, "legs": legs, "stake_usd": round(stake, 2), "max_loss_pct": float(order.get("max_loss_pct", 0.02)), "last_quote_underlying": round(last, 4), "thesis": order.get("thesis", "")[:500], "status": "open", "mode": "live" if live_mode() else "dry_run", } if live_mode() and os.environ.get("ITF_OPTIONS_LIVE", "").lower() in ("1","true","yes"): # Alpaca multi-leg options (order_class=mleg) — canonical alpaca-py pattern. # Single atomic POST replaces the old per-leg loop which broke spread pricing # AND left naked legs when one fill succeeded and another failed. try: import requests key = os.environ["ALPACA_PAPER_KEY"] secret = os.environ["ALPACA_PAPER_SECRET"] headers = {"APCA-API-KEY-ID": key, "APCA-API-SECRET-KEY": secret} if len(legs) == 1: # Single-leg: simple market order on OCC symbol (no mleg). leg = legs[0] payload = { "symbol": leg["symbol"], "qty": leg["qty"], "side": leg["side"], "type": "market", "time_in_force": "day", } r = requests.post("https://paper-api.alpaca.markets/v2/orders", headers=headers, json=payload, timeout=10) r.raise_for_status() entry["broker_order_ids"] = [r.json().get("id")] else: # Multi-leg: one mleg order, 2-4 legs. ratio_qty defines leg proportions; # position_intent declares open vs close so Alpaca can validate margin. mleg_legs = [] for leg in legs: mleg_legs.append({ "symbol": leg["symbol"], "side": leg["side"], "ratio_qty": str(leg["qty"]), "position_intent": "buy_to_open" if leg["side"] == "buy" else "sell_to_open", }) payload = { "order_class": "mleg", "qty": "1", "type": "market", "time_in_force": "day", "legs": mleg_legs, } r = requests.post("https://paper-api.alpaca.markets/v2/orders", headers=headers, json=payload, timeout=10) r.raise_for_status() resp = r.json() entry["broker_order_ids"] = [resp.get("id")] entry["broker_class"] = "mleg" entry["broker_legs_count"] = len(mleg_legs) except Exception as e: entry["status"] = "broker_error" # Capture Alpaca body text when available (RequestException.response) so we can # diagnose broker_errors beyond the generic "422 Unprocessable Entity" prefix. body = "" try: body = " | body=" + getattr(e, "response", None).text[:400] # type: ignore except Exception: pass entry["error"] = (str(e) + body)[:600] # 2026-04-25 — mirror reject into agent_ledger (options path) _append_ledger({ "tid": agent_tid, "event": "broker_reject", "ticker": underlying, "side": entry.get("side"), "stake": entry.get("stake_usd"), "instrument": "option", "reason_code": _classify_reject(entry["error"]), "reason": entry["error"][:300], }) else: entry["sim_opened_at_underlying"] = last positions.setdefault(agent_tid, []).append(entry) _save_positions(positions) _append_order_log(entry) # v2.5 — reserve option stake from agent's sub-bankroll if entry.get("status") == "open": new_bal = reserve_bankroll(agent_tid, stake, meta={ "underlying": underlying, "strategy": strategy, "stake": round(stake, 2), "instrument": "option", }) entry["agent_bankroll_after_reserve"] = new_bal return entry def _mark_to_market(p: Dict[str, Any], quote_fn) -> Dict[str, Any]: """Compute realized P&L on close. Mutates `p` with realized_pnl_usd + exit_price + return_pct. For equities: P&L = qty * (exit - entry) for long, qty * (entry - exit) for short. For options (dry-run): approximate intrinsic value delta via underlying quote — coarse but gives a direction/magnitude signal. Live options P&L pulled from broker fill feed. For broker_error / rejected entries: leave P&L at 0.0. """ if p.get("status") in ("broker_error", "rejected"): p["realized_pnl_usd"] = 0.0 return p if p.get("instrument_type") == "option": # Dry-run option P&L: Δ(underlying) × stake × direction sign. Not real Greeks, # but enough to produce non-zero leaderboard numbers and catch obviously # losing theses. Live mode will overwrite via broker fills. underlying_entry = float(p.get("last_quote_underlying") or 0) or 0.0 exit_q = quote_fn(p.get("underlying", "")) or underlying_entry if underlying_entry > 0: delta_pct = (exit_q - underlying_entry) / underlying_entry else: delta_pct = 0.0 direction = 1 if p.get("option_type", "call") == "call" else -1 if p.get("strategy") in ("vertical_credit",): direction *= -1 # credit: we profit on small moves, lose on large in direction stake = float(p.get("stake_usd") or 0) pnl = stake * delta_pct * direction # Cap loss at stake (long premium) — no naked unlimited here by design. pnl = max(pnl, -stake) p["realized_pnl_usd"] = round(pnl, 2) p["exit_underlying"] = round(exit_q, 4) p["return_pct"] = round(delta_pct * direction, 5) return p entry_px = float(p.get("entry_price") or 0) or 0.0 qty = float(p.get("qty") or 0) or 0.0 ticker = p.get("ticker", "") exit_px = quote_fn(ticker) or entry_px if p.get("side") == "short": gross = qty * (entry_px - exit_px) else: gross = qty * (exit_px - entry_px) p["realized_pnl_usd"] = round(gross, 2) p["exit_price"] = round(exit_px, 4) if entry_px > 0: p["return_pct"] = round((exit_px - entry_px) / entry_px, 5) else: p["return_pct"] = 0.0 return p def close_expired(now_utc: datetime, quote_fn=None) -> List[Dict[str, Any]]: """Walk positions and close any that passed persona.max_hold or hit EOD flatten. `quote_fn(ticker) -> last_price` is injected from app.py so we can stay decoupled from the quote_bus import. If None, we fall back to entry_price (zero P&L) but still mark status=closed_expired. """ closed: List[Dict[str, Any]] = [] positions = _load_positions() qf = quote_fn or (lambda _t: None) for agent_tid, rows in positions.items(): for p in rows: if p.get("status") != "open": continue try: opened = datetime.fromisoformat(p["ts"].replace("Z", "+00:00")) except Exception: continue age_min = (now_utc - opened).total_seconds() / 60.0 eod = (now_utc.hour > EOD_FLATTEN_UTC_HOUR or (now_utc.hour == EOD_FLATTEN_UTC_HOUR and now_utc.minute >= EOD_FLATTEN_UTC_MIN)) # We don't know per-agent max_hold here without loading personas; use 240 as a ceiling. if age_min > 240 or eod: p["status"] = "closed_expired" p["closed_at"] = now_utc.strftime("%Y-%m-%dT%H:%M:%SZ") _mark_to_market(p, qf) # v2.5 — credit stake + realized_pnl back to agent's sub-bankroll _stake = float(p.get("stake_usd") or 0) _pnl = float(p.get("realized_pnl_usd") or 0) credit_bankroll(agent_tid, _stake + _pnl, meta={ "ticker": p.get("ticker") or p.get("underlying"), "event_type": "eod_or_expired_close", "stake_returned": round(_stake, 2), "realized_pnl": round(_pnl, 2), }) closed.append(p) _save_positions(positions) return closed def pnl_snapshot(quote_fn=None) -> Dict[str, Any]: """Per-agent P&L aggregate. Realized = sum(realized_pnl_usd) on closed rows. Unrealized = mark-to-market on still-open rows. Total = realized + unrealized. Uses `quote_fn(ticker)` for the open-position mark. If quote missing, mark is 0. """ positions = _load_positions() qf = quote_fn or (lambda _t: None) per_agent: Dict[str, Dict[str, float]] = {} for agent_tid, rows in positions.items(): realized = 0.0 unrealized = 0.0 wins = 0 losses = 0 trades_closed = 0 trades_open = 0 for p in rows: if p.get("status") == "open": trades_open += 1 # Shadow-copy to mark-to-market without persisting. tmp = dict(p) _mark_to_market(tmp, qf) unrealized += float(tmp.get("realized_pnl_usd") or 0) elif p.get("status", "").startswith("closed"): trades_closed += 1 r = float(p.get("realized_pnl_usd") or 0) realized += r if r > 0: wins += 1 elif r < 0: losses += 1 per_agent[agent_tid] = { "realized_pnl_usd": round(realized, 2), "unrealized_pnl_usd": round(unrealized, 2), "total_pnl_usd": round(realized + unrealized, 2), "trades_closed": trades_closed, "trades_open": trades_open, "wins": wins, "losses": losses, "win_rate": round(wins / trades_closed, 4) if trades_closed else 0.0, } total_realized = sum(a["realized_pnl_usd"] for a in per_agent.values()) total_unrealized = sum(a["unrealized_pnl_usd"] for a in per_agent.values()) return { "per_agent": per_agent, "fleet_realized_pnl_usd": round(total_realized, 2), "fleet_unrealized_pnl_usd": round(total_unrealized, 2), "fleet_total_pnl_usd": round(total_realized + total_unrealized, 2), } MIN_HOLD_SEC = int(os.environ.get("ITF_MIN_HOLD_SEC", "180")) # 3 min default. 2026-04-25: push fleet from ~60 to ~500 fills/day. Was 900 (15min) anti-churn, but BP drain root-cause was order pile-up (fixed via MAX_PENDING_PER_SYMBOL=4 + cancel_stale_pending). Account equity > $25K so PDT 4-roundtrip rule is moot; daytrading_buying_power=0 is the structural Reg-T limit (margin used). MIN_HOLD now prevents only sub-3min flap. # ───── 2026-04-22 BROKER-FILL RECONCILIATION ───── # The submit path reserves stake from the agent's sub-bankroll, but until # 2026-04-22 no code path credited realized PnL back from Alpaca fills, so every # /api/bankrolls read still showed the cold-start seed. reconcile_broker_fills() # polls Alpaca /v2/account/activities/FILL, matches fills to local positions via # broker_order_id, and credits realized_pnl to the right agent on closing sides. # A cursor at data/intraday/fill_reconciliation_cursor.json prevents double-count. def _load_recon_cursor() -> Dict[str, Any]: if not RECON_CURSOR_PATH.exists(): return {"seen_ids": [], "last_run_at": None} try: d = json.loads(RECON_CURSOR_PATH.read_text()) if not isinstance(d, dict): return {"seen_ids": [], "last_run_at": None} d.setdefault("seen_ids", []) d.setdefault("last_run_at", None) return d except Exception: return {"seen_ids": [], "last_run_at": None} def _save_recon_cursor(cur: Dict[str, Any]) -> None: # Keep seen_ids bounded — 2k most-recent is plenty for a ~15min lookback. seen = cur.get("seen_ids") or [] if len(seen) > 2000: cur["seen_ids"] = seen[-2000:] RECON_CURSOR_PATH.parent.mkdir(parents=True, exist_ok=True) RECON_CURSOR_PATH.write_text(json.dumps(cur, indent=2, default=str)) global _LEDGER_DIRTY _LEDGER_DIRTY = True def _fetch_fill_activities(lookback_min: int) -> List[Dict[str, Any]]: """GET /v2/account/activities/FILL?after=&direction=desc — stdlib only.""" if not live_mode(): return [] key = os.environ.get("ALPACA_PAPER_KEY") secret = os.environ.get("ALPACA_PAPER_SECRET") if not (key and secret): return [] import urllib.parse import urllib.request after = (datetime.now(timezone.utc) - timedelta(minutes=max(1, lookback_min))).strftime( "%Y-%m-%dT%H:%M:%SZ" ) qs = urllib.parse.urlencode({ "activity_types": "FILL", "after": after, "direction": "desc", "page_size": "100", }) url = f"https://paper-api.alpaca.markets/v2/account/activities?{qs}" req = urllib.request.Request( url, headers={"APCA-API-KEY-ID": key, "APCA-API-SECRET-KEY": secret} ) try: with urllib.request.urlopen(req, timeout=10) as r: # noqa: S310 body = r.read().decode("utf-8") data = json.loads(body or "[]") return data if isinstance(data, list) else [] except Exception: return [] def _index_positions_by_order_id( positions: Dict[str, List[Dict[str, Any]]] ) -> Dict[str, Dict[str, Any]]: """Build {broker_order_id -> position_ref} + {client_order_id -> position_ref}. position_ref is a dict {"agent_tid": ..., "position": }.""" idx: Dict[str, Dict[str, Any]] = {} for agent_tid, rows in (positions or {}).items(): for p in (rows or []): oid = p.get("broker_order_id") cid = p.get("client_order_id") if oid: idx[str(oid)] = {"agent_tid": agent_tid, "position": p} if cid: idx[str(cid)] = {"agent_tid": agent_tid, "position": p} return idx def reconcile_broker_fills(lookback_min: int = 15) -> Dict[str, Any]: """Poll Alpaca FILL activities and credit realized PnL back to per-agent sub-bankrolls on closing fills (FIFO within a matched position). Called at the top of every tick so executor.get_bankroll(tid) reflects true post-fill balance before the next prompt is built. Behavior: * In dry-run mode, no-op (returns zeroed stats). * Fills already in cursor.seen_ids are skipped. * A fill whose order_id (or client_order_id) matches a local open position with an OPPOSITE broker side is treated as a CLOSE: - computes realized_pnl = qty_closed * (fill_px - entry_px) * direction - credits (stake_portion + realized_pnl) to the agent's bankroll - marks the position status="closed_by_fill" when the full qty closed A same-side fill is just an open-fill confirmation — updates filled_avg_price/filled_qty on the position, no bankroll move (the stake was already reserved at submit). Returns a stats dict: { "fills_processed": N, "closes_applied": K, "bankroll_delta_by_agent": {tid: float_delta_usd}, "unmatched_fills": M, "skipped_seen": S, "mode": "live" | "dry_run", } """ stats: Dict[str, Any] = { "fills_processed": 0, "closes_applied": 0, "bankroll_delta_by_agent": {}, "unmatched_fills": 0, "skipped_seen": 0, "mode": "live" if live_mode() else "dry_run", } if not live_mode(): return stats cursor = _load_recon_cursor() seen: List[str] = list(cursor.get("seen_ids") or []) seen_set = set(seen) fills = _fetch_fill_activities(lookback_min) if not fills: cursor["last_run_at"] = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") _save_recon_cursor(cursor) return stats positions = _load_positions() idx = _index_positions_by_order_id(positions) dirty = False # Alpaca returns fills newest-first when direction=desc; process oldest-first # so FIFO closes are applied in trade order. for fill in reversed(fills): fill_id = str(fill.get("id") or "") if not fill_id: continue if fill_id in seen_set: stats["skipped_seen"] += 1 continue seen_set.add(fill_id) seen.append(fill_id) order_id = str(fill.get("order_id") or "") client_order_id = str(fill.get("client_order_id") or "") symbol = (fill.get("symbol") or "").upper() fill_side = (fill.get("side") or "").lower() # buy | sell | sell_short try: qty_filled = float(fill.get("qty") or 0) except Exception: qty_filled = 0.0 try: fill_px = float(fill.get("price") or 0) except Exception: fill_px = 0.0 match = idx.get(order_id) or idx.get(client_order_id) stats["fills_processed"] += 1 if not match: # Fill we don't know about (e.g. bracket child stop-loss legs Alpaca # generates internally). Log but don't credit — we can't safely pick # an agent without a position link. stats["unmatched_fills"] += 1 _append_ledger({ "event": "unmatched_fill", "source": "broker_reconcile", "fill_id": fill_id, "order_id": order_id, "client_order_id": client_order_id, "ticker": symbol, "side": fill_side, "qty": qty_filled, "price": fill_px, }) continue agent_tid = match["agent_tid"] pos = match["position"] pos_side = (pos.get("side") or "").lower() # long | short # Broker "sell" (for a long) OR "buy" (for a short) = closing fill. is_close = ( (pos_side == "long" and fill_side == "sell") or (pos_side == "short" and fill_side in ("buy", "buy_to_cover")) ) is_open_confirm = ( (pos_side == "long" and fill_side == "buy") or (pos_side == "short" and fill_side in ("sell", "sell_short")) ) if is_open_confirm: # Mark the fill on the position so subsequent MTM reads are honest. pos["filled_avg_price"] = fill_px or pos.get("filled_avg_price") # Accumulate filled_qty across partial fills. prev_filled = float(pos.get("filled_qty") or 0) pos["filled_qty"] = round(prev_filled + qty_filled, 6) pos["filled_at"] = fill.get("transaction_time") or pos.get("filled_at") if (pos.get("broker_status") or "").lower() != "filled": pos["broker_status"] = "filled" if pos["filled_qty"] >= float(pos.get("qty") or 0) else "partially_filled" dirty = True _append_ledger({ "event": "open_fill_confirm", "source": "broker_reconcile", "fill_id": fill_id, "ts": fill.get("transaction_time"), "agent_tid": agent_tid, "ticker": symbol, "side": fill_side, "qty": qty_filled, "price": fill_px, "realized_pnl": 0.0, }) continue if not is_close: # Side doesn't make sense vs our recorded pos_side (e.g. recorded as # long but fill came back as sell_short). Log and skip — surfaces as # unmatched for audit, cursor still advances. stats["unmatched_fills"] += 1 _append_ledger({ "event": "side_mismatch", "source": "broker_reconcile", "fill_id": fill_id, "agent_tid": agent_tid, "pos_side": pos_side, "fill_side": fill_side, "ticker": symbol, }) continue # CLOSING FILL: credit stake_portion + realized_pnl to the agent. entry_px = float(pos.get("entry_price") or pos.get("filled_avg_price") or 0) total_qty = float(pos.get("qty") or 0) or qty_filled if total_qty <= 0: stats["unmatched_fills"] += 1 continue portion = min(1.0, qty_filled / total_qty) if total_qty else 1.0 stake = float(pos.get("stake_usd") or 0) stake_portion = round(stake * portion, 2) if pos_side == "long": pnl = qty_filled * (fill_px - entry_px) else: pnl = qty_filled * (entry_px - fill_px) pnl = round(pnl, 2) credit = stake_portion + pnl # Don't double-credit: if close_expired/close_position already credited # this position (status starts with "closed"), skip the bankroll write # but still ledger the broker fill for audit. already_closed = str(pos.get("status") or "").startswith("closed") if not already_closed: credit_bankroll(agent_tid, credit, meta={ "event_type": "broker_reconcile_close", "source": "broker_reconcile", "fill_id": fill_id, "ticker": symbol, "qty_closed": qty_filled, "fill_price": fill_px, "entry_price": entry_px, "stake_portion_returned": stake_portion, "realized_pnl": pnl, }) stats["bankroll_delta_by_agent"][agent_tid] = round( stats["bankroll_delta_by_agent"].get(agent_tid, 0.0) + credit, 2 ) # Mark position closed when the full qty has been sold off. if qty_filled >= total_qty - 1e-6: pos["status"] = "closed_by_fill" pos["closed_at"] = fill.get("transaction_time") or datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") pos["realized_pnl_usd"] = pnl pos["exit_price"] = fill_px dirty = True _append_ledger({ "event": "close_fill", "source": "broker_reconcile", "fill_id": fill_id, "ts": fill.get("transaction_time"), "agent_tid": agent_tid, "ticker": symbol, "side": fill_side, "qty": qty_filled, "price": fill_px, "entry_price": entry_px, "realized_pnl": pnl, "stake_portion_returned": stake_portion, "already_closed_locally": already_closed, }) stats["closes_applied"] += 1 if dirty: _save_positions(positions) cursor["seen_ids"] = seen cursor["last_run_at"] = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") _save_recon_cursor(cursor) return stats def close_position(agent_tid: str, ticker: str) -> Dict[str, Any]: """2026-04-21 — agent-driven close. Mark matching local open positions closed and (in live mode) submit Alpaca DELETE /v2/positions/{symbol} to flatten the broker position. Returns entry-style dict mirroring submit(). Scope: matches ALL open positions for this agent_tid + ticker pair. Broker close is market-time-in-force, so crypto closes GTC via order, equities via the dedicated positions-close endpoint (net flat). 2026-04-22 — MIN_HOLD_SEC guard: if ALL matched positions are younger than MIN_HOLD_SEC, reject the close with status=blocked_by_min_hold. Forces the fleet to hold through intra-hour noise instead of churning daytrades. """ positions = _load_positions() ticker_u = (ticker or "").upper().strip() matched: List[Dict[str, Any]] = [ p for p in positions.get(agent_tid, []) if p.get("status") == "open" and (p.get("ticker", "") or "").upper().strip() == ticker_u ] if matched and MIN_HOLD_SEC > 0: now_utc = datetime.now(timezone.utc) eligible = [] for p in matched: try: opened = datetime.fromisoformat((p.get("opened_at") or p.get("ts") or "").replace("Z", "+00:00")) if (now_utc - opened).total_seconds() >= MIN_HOLD_SEC: eligible.append(p) except Exception: eligible.append(p) # if timestamp unreadable, don't block if not eligible: blocked = { "ts": now_utc.strftime("%Y-%m-%dT%H:%M:%SZ"), "agent_tid": agent_tid, "ticker": ticker_u, "action": "close_position", "status": "blocked_by_min_hold", "min_hold_sec": MIN_HOLD_SEC, "youngest_age_sec": int(min( (now_utc - datetime.fromisoformat((p.get("opened_at") or p.get("ts") or "").replace("Z", "+00:00"))).total_seconds() for p in matched )), "matched_positions": len(matched), } _append_order_log(blocked) return blocked matched = eligible entry = { "ts": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"), "agent_tid": agent_tid, "ticker": ticker_u, "action": "close_position", "matched_positions": len(matched), "mode": "live" if live_mode() else "dry_run", } if not matched: entry["status"] = "no_open_position" _append_order_log(entry) return entry if live_mode(): import requests key = os.environ["ALPACA_PAPER_KEY"] secret = os.environ["ALPACA_PAPER_SECRET"] headers = {"APCA-API-KEY-ID": key, "APCA-API-SECRET-KEY": secret} try: r = requests.delete( f"https://paper-api.alpaca.markets/v2/positions/{ticker_u}", headers=headers, timeout=10, ) if r.ok: entry["broker_status"] = "closed" entry["broker_resp"] = (r.json() if r.text else {}).get("status", "submitted") else: entry["broker_status"] = f"error_{r.status_code}" entry["broker_resp"] = r.text[:300] except Exception as e: entry["broker_status"] = "exception" entry["broker_resp"] = str(e)[:300] # Mark all matched local positions closed + credit stake back to sub-bankroll. # For agent-driven closes we don't have a live quote for exact P&L, so credit # only the reserved stake (P&L reconciles from broker fills → next tick via # a snapshot reconciliation). This is intentionally conservative: the stake # returns, any unrealized gain is "free" on close until reconciled. total_stake_returned = 0.0 for p in matched: p["status"] = "closed_by_agent" p["closed_at"] = entry["ts"] total_stake_returned += float(p.get("stake_usd") or 0) if total_stake_returned > 0: credit_bankroll(agent_tid, total_stake_returned, meta={ "ticker": ticker_u, "event_type": "agent_close", "stake_returned": round(total_stake_returned, 2), "n_positions": len(matched), }) _save_positions(positions) _append_order_log(entry) return entry def read_trades(limit: int = 200) -> List[Dict[str, Any]]: """Tail the dry_run_orders.jsonl log. Shape: every submit() call (fill or reject).""" if not ORDERS_JSONL.exists(): return [] rows: List[Dict[str, Any]] = [] try: lines = ORDERS_JSONL.read_text().splitlines() except Exception: return [] for line in lines[-limit:]: try: rows.append(json.loads(line)) except Exception: continue return rows def list_open() -> List[Dict[str, Any]]: positions = _load_positions() out: List[Dict[str, Any]] = [] for agent_tid, rows in positions.items(): for p in rows: if p.get("status") == "open": out.append(p) return out # ───── 2026-04-22 — HF-persistence for the 4 ledger files ───── # HF Spaces wipe /app on every factory_reboot. positions.json, agent_bankrolls.json, # fill_reconciliation_cursor.json and agent_ledger.jsonl live under /app/data/intraday # → every restart = full attribution reset (IA confirmed "36 hours lost"). # Solution: persist_ledgers_to_hub() uploads all four files to the ITF repo itself # (path_in_repo=data/intraday/*). restore_ledgers.py (shipped alongside app.py in # the Dockerfile) downloads them on boot before uvicorn starts. _ITF_REPO_ID = os.environ.get("SPACE_ID") or "LBJLincoln26/intraday-trading-floor" def _hf_token() -> Optional[str]: """Prefer HF_TOKEN_2 (write token, what the memory index says to use), then NBA token, then generic HF_TOKEN. Never raise — silent skip if absent.""" for k in ("HF_TOKEN_2", "HF_TOKEN_NBA", "HF_WRITE_TOKEN", "NOMOS_HF_TOKEN", "HF_TOKEN"): v = os.environ.get(k) if v: return v return None def persist_ledgers_to_hub(force: bool = False) -> Dict[str, Any]: """Upload the 4 ledger files to the ITF HF repo so a factory_reboot can re-hydrate them via restore_ledgers.py. One commit per invocation (batched via `create_commit` with up to 4 ops) so we don't spam the repo with tick-cadence commits. Called at the end of every tick_once(); _LEDGER_DIRTY gates so ticks with no mutations skip the Hub round-trip entirely. Rules: * missing file → skipped (no error). * agent_ledger.jsonl → skipped when size > 5 MB (append-only, would thrash the LFS-free 10 MB quota). * no HF token → returns early with {"skipped": "no-token"}. * any exception → caught, returned in `errors`, never raised. Returns {"uploaded": [...], "skipped": "...", "errors": [...]}. """ out: Dict[str, Any] = {"uploaded": [], "errors": []} global _LEDGER_DIRTY if not force and not _LEDGER_DIRTY: out["skipped"] = "clean" return out tok = _hf_token() if not tok: out["skipped"] = "no-token" return out try: from huggingface_hub import HfApi from huggingface_hub import CommitOperationAdd except Exception as e: out["errors"].append(f"import_hfapi: {str(e)[:200]}") return out candidates = [ (POSITIONS_PATH, "data/intraday/positions.json", False), (BANKROLLS_PATH, "data/intraday/agent_bankrolls.json", False), (RECON_CURSOR_PATH, "data/intraday/fill_reconciliation_cursor.json", False), (LEDGER_JSONL, "data/intraday/agent_ledger.jsonl", True), ] # 2026-04-28 — also persist last 7 days of per-day decisions/*.jsonl so the # audit pipeline (per_agent_deep_audit.py) keeps the rationale trail across # factory_reboot. Without this every reset wiped /api/decisions history, # making cross-day comparison impossible. Each file is small (~17 rows/day # × ~2 KB/row = 30 KB/day), capped at 7 days = ~200 KB total. try: decisions_dir = REPO / "data" / "intraday" / "decisions" if decisions_dir.is_dir(): from datetime import datetime as _dt, timezone as _tz, timedelta as _td today = _dt.now(_tz.utc).date() for i in range(7): d = today - _td(days=i) p = decisions_dir / f"{d.isoformat()}.jsonl" if p.exists(): candidates.append( (p, f"data/intraday/decisions/{d.isoformat()}.jsonl", True) ) except Exception as _e: out["errors"].append(f"decisions_enum: {str(_e)[:200]}") ops: List[Any] = [] for local, remote, is_ledger in candidates: try: if not local.exists(): continue if is_ledger: try: size = local.stat().st_size except Exception: size = 0 if size > _LEDGER_MAX_UPLOAD_BYTES: out.setdefault("skipped_big", []).append( {"path": remote, "bytes": size} ) continue ops.append(CommitOperationAdd(path_in_repo=remote, path_or_fileobj=str(local))) except Exception as e: out["errors"].append(f"{remote}: {str(e)[:200]}") if not ops: out["skipped"] = "no-ops" _LEDGER_DIRTY = False return out try: api = HfApi(token=tok) ts = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") api.create_commit( repo_id=_ITF_REPO_ID, repo_type="space", operations=ops, commit_message=f"[ITF-LEDGER] tick snapshot {ts}", ) out["uploaded"] = [op.path_in_repo for op in ops] _LEDGER_DIRTY = False except Exception as e: out["errors"].append(f"commit: {str(e)[:300]}") return out