File: //opt/trading-bot/runner.py
import asyncio
import logging
from datetime import datetime, timezone
from config import settings
from db import SessionLocal
from models import MarketSnapshot, PortfolioState, Position, PnlHistory
from strategy import MultiAssetBasisStrategy
from allocator import Allocator
from risk_engine import RiskEngine
from execution import ExecutionEngine
from optimizer import optimizer
from telemetry import strategy_status
from market_hours import is_market_open
logger = logging.getLogger(__name__)
class TradingRunner:
def __init__(self, strategy_name: str = "MultiAssetAlpha", strategy_instance = None):
self.strategy_name = strategy_name
self.strategy = strategy_instance or MultiAssetBasisStrategy()
self.risk_engine = RiskEngine()
self.allocator = Allocator(self.risk_engine)
self.execution = ExecutionEngine()
self.is_running = False
async def run_loop(self):
self.is_running = True
logger.info(f"Starting Trading Runner for {self.strategy_name} (Interval: {settings.STRATEGY_INTERVAL}s)")
strategy_status[self.strategy_name] = {"message": "Initializing...", "color": "var(--text-secondary)"}
while self.is_running:
await self.run_once()
await asyncio.sleep(settings.STRATEGY_INTERVAL)
async def run_once(self):
"""Executes a single iteration of the strategy logic."""
try:
if not self.risk_engine.is_trading_allowed(self.strategy_name):
logger.warning(f"Risk engine blocked trading cycle for {self.strategy_name}.")
strategy_status[self.strategy_name] = {"message": "BLOCKED: Risk Engine", "color": "var(--danger)"}
return
# Market Guard for strategies that trade equities/ETFs
# These strategies handle market hours internally via their compute_signals()
# but we add a safety guard here too
equity_strategies = ["PairsArbitrage"] # WhaleTracker/FearGreed/LiquidationCascade are crypto-only, FundingHarvester is crypto-only
if self.strategy_name in equity_strategies:
is_open, reason = is_market_open()
if not is_open:
strategy_status[self.strategy_name] = {"message": f"HIBERNATING: Market {reason}", "color": "var(--text-secondary)"}
return
with SessionLocal() as session:
# 1. Get current state for this strategy
portfolio = self.execution.get_portfolio_state(session, strategy_name=self.strategy_name)
positions = self.execution.get_positions(session, strategy_name=self.strategy_name)
# 2. Recalculate exposure and EQUITY for this strategy
market_value = 0.0
current_exposure = 0.0
unrealized_pnl = 0.0
all_tickers = session.query(MarketSnapshot).order_by(MarketSnapshot.ts_ingested.desc()).limit(1200).all()
market_map = {t.symbol: t for t in all_tickers}
for symbol, qty in positions.items():
if symbol in market_map:
price = float(market_map[symbol].last)
market_value += float(qty) * price
current_exposure += abs(float(qty)) * price
pos_record = session.query(Position).filter_by(symbol=symbol, strategy_name=self.strategy_name).first()
if pos_record:
unrealized_pnl += (price - float(pos_record.avg_price)) * float(qty)
latest_state = session.query(PortfolioState).filter_by(strategy_name=self.strategy_name).order_by(PortfolioState.ts.desc()).first()
current_cash = float(latest_state.cash) if latest_state else 10000.0
# Equity = Cash + Value of all positions
current_equity = current_cash + market_value
# === PROPER DRAWDOWN TRACKING ===
# Find peak equity from history
from sqlalchemy import func as sa_func
peak_row = session.query(sa_func.max(PortfolioState.equity)).filter_by(strategy_name=self.strategy_name).scalar()
peak_equity = float(peak_row) if peak_row else 10000.0
peak_equity = max(peak_equity, current_equity) # Update peak if new high
current_drawdown = ((peak_equity - current_equity) / peak_equity * 100) if peak_equity > 0 else 0.0
state_update = PortfolioState(
strategy_name=self.strategy_name,
equity=current_equity,
cash=current_cash,
exposure=current_exposure,
drawdown=current_drawdown,
ts=datetime.now(timezone.utc)
)
session.add(state_update)
# === NEGATIVE BALANCE GUARD ===
if current_cash <= 0:
logger.warning(f"{self.strategy_name}: Cash depleted (${current_cash:.2f}). Skipping signals.")
strategy_status[self.strategy_name] = {"message": f"BLOCKED: No Cash (${current_cash:.0f})", "color": "var(--danger)"}
session.commit()
return
# 3. Compute Signals
raw_signals = self.strategy.compute_signals()
if not raw_signals:
# Use strategy diagnostic for detailed status if available
diag = getattr(self.strategy, 'get_diagnostic', lambda: None)()
if diag:
strategy_status[self.strategy_name] = {"message": f"WAITING: {diag}", "color": "var(--text-secondary)"}
else:
strategy_status[self.strategy_name] = {"message": "WAITING: No Entry Signals", "color": "var(--text-secondary)"}
session.commit()
return
# Apply Optimizer
signals = optimizer.apply_convection_filter(raw_signals, self.strategy_name)
if not signals:
strategy_status[self.strategy_name] = {"message": "CAUTION: Optimizer Filtered", "color": "var(--warning)"}
session.commit()
return
# === POSITION ACCUMULATION GUARD ===
# Don't add new LONG positions if we already hold that symbol
# This prevents the same signal from stacking positions each cycle
filtered_signals = []
for signal in signals:
symbol = signal.get('symbol') or f"{signal.get('asset', 'BTC')}/USDC"
current_qty = positions.get(symbol, 0.0)
action = signal.get('action', 'NEUTRAL')
if action == 'CLOSE':
# Always allow closes
filtered_signals.append(signal)
elif action == 'LONG' and current_qty > 1e-8:
# Already long this symbol — skip
logger.debug(f"{self.strategy_name}: Skipping LONG {symbol}, already holding {current_qty:.6f}")
continue
elif action == 'SHORT' and current_qty < -1e-8:
# Already short — skip
logger.debug(f"{self.strategy_name}: Skipping SHORT {symbol}, already short {current_qty:.6f}")
continue
else:
filtered_signals.append(signal)
signals = filtered_signals
if not signals:
strategy_status[self.strategy_name] = {"message": "Active (holding)", "color": "var(--success)"}
session.commit()
return
# Cap total positions per strategy
MAX_POSITIONS = 3
active_positions = sum(1 for q in positions.values() if abs(q) > 1e-8)
available_slots = max(0, MAX_POSITIONS - active_positions)
new_entries = [s for s in signals if s.get('action') != 'CLOSE']
closes = [s for s in signals if s.get('action') == 'CLOSE']
signals = closes + new_entries[:available_slots]
# 4. Generate and Execution Target Allocations
for signal in signals:
# Resolve symbol
asset = signal.get('asset', 'BTC')
symbol = signal.get('symbol') or f"{asset}/USDC"
target_qty = self.allocator.get_target_qty(signal, current_equity, available_cash=current_cash)
current_qty = positions.get(symbol, 0.0)
if abs(target_qty - current_qty) > 1e-8:
self.execution.execute_paper_trade(
session,
symbol,
target_qty,
current_qty,
price=signal['price'],
strategy_name=self.strategy_name
)
session.commit()
strategy_status[self.strategy_name] = {"message": "Active", "color": "var(--success)"}
except Exception as e:
logger.error(f"Error in TradingRunner.{self.strategy_name}.run_once: {e}", exc_info=True)
strategy_status[self.strategy_name] = {"message": f"CRASH: {str(e)[:20]}...", "color": "var(--danger)"}
def stop(self):
self.is_running = False