from typing import Dict, List, Optional, Tuple, Callable import threading import time import math import config def pearson_correlation(x: List[float], y: List[float]) -> float: """Compute Pearson correlation coefficient r between two series. r = sum((x - x̄)(y - ȳ)) / sqrt(sum(x - x̄)^2 * sum(y - ȳ)^2) Returns value in [-1, 1]. |r| > 0.75 indicates strong correlation. """ n = min(len(x), len(y)) if n < 3: return 0.0 x, y = x[:n], y[:n] x_mean = sum(x) / n y_mean = sum(y) / n num = sum((xi - x_mean) * (yi - y_mean) for xi, yi in zip(x, y)) den_x = math.sqrt(sum((xi - x_mean) ** 2 for xi in x)) den_y = math.sqrt(sum((yi - y_mean) ** 2 for yi in y)) if den_x == 0 or den_y == 0: return 0.0 r = num / (den_x * den_y) return max(-1.0, min(1.0, r)) class PositionSizer: """Calculates position size based on risk and confluence strength.""" def __init__(self, account_balance: float = 10000.0, risk_per_trade: float = 0.01): self.account_balance = account_balance self.risk_per_trade = risk_per_trade self.positions = {} def calculate_position_size( self, pair: str, confluence_strength: float, entry_price: float, stop_loss_pips: float = 50, bid: float = None, ask: float = None, spread: float = None, ) -> float: risk_amount = self.account_balance * self.risk_per_trade confidence_multiplier = confluence_strength / 100.0 pip_value_per_lot = 10.0 max_loss_per_lot = stop_loss_pips * pip_value_per_lot if max_loss_per_lot > 0: base_position = risk_amount / max_loss_per_lot position_size = base_position * confidence_multiplier else: position_size = 0.0 # Spread penalty (Task 2.1): wide spreads reduce size by up to 20% if spread is not None and spread > 0: spread_penalty = min(spread * 100, 0.2) # cap at 20% penalty position_size *= (1.0 - spread_penalty) position_size = max(0.01, min(position_size, 5.0)) return position_size def add_position(self, pair: str, position_size: float, entry_price: float): self.positions[pair] = { 'size': position_size, 'entry_price': entry_price, 'status': 'OPEN' } def close_position(self, pair: str, exit_price: float) -> Optional[Dict]: if pair not in self.positions: return None pos = self.positions[pair] price_delta = exit_price - pos['entry_price'] pl = price_delta * pos['size'] * 100000 result = { 'pair': pair, 'entry': pos['entry_price'], 'exit': exit_price, 'size': pos['size'], 'pnl': pl, 'pnl_pips': price_delta * 10000 } del self.positions[pair] return result class GridHedging: """Grid hedging system for drawdown protection.""" def __init__(self, grid_levels: int = 3): self.grid_levels = grid_levels self.hedges = [] def create_hedge_grid( self, pair: str, entry_price: float, position_size: float, grid_distance: float = 0.50 ) -> List[Dict]: self.hedges = [] if self.grid_levels < 2: return self.hedges hedge_size = position_size * 0.5 / (self.grid_levels - 1) for level in range(1, self.grid_levels): hedge_price = entry_price - (grid_distance * level / 10000) self.hedges.append({ 'pair': pair, 'level': level, 'price': hedge_price, 'size': hedge_size, 'type': 'HEDGE' }) return self.hedges def get_total_hedge_exposure(self) -> float: return sum(h['size'] for h in self.hedges) def get_hedges_for_pair(self, pair: str) -> List[Dict]: return [h for h in self.hedges if h['pair'] == pair] class PortfolioExposure: """Manage portfolio-level exposure and leverage.""" def __init__(self, max_portfolio_leverage: float = 2.0): self.max_leverage = max_portfolio_leverage self.positions = {} self.total_exposure = 0.0 def can_add_position(self, position_size: float, account_balance: float) -> bool: new_exposure = self.total_exposure + position_size max_exposure = account_balance * self.max_leverage return new_exposure <= max_exposure def add_position(self, pair: str, position_size: float): self.positions[pair] = position_size self.total_exposure = sum(self.positions.values()) def remove_position(self, pair: str): if pair in self.positions: del self.positions[pair] self.total_exposure = sum(self.positions.values()) def get_leverage_ratio(self, account_balance: float) -> float: if account_balance <= 0: return 0.0 return self.total_exposure / account_balance def get_exposure_percentage(self, pair: str) -> float: if self.total_exposure <= 0: return 0.0 return (self.positions.get(pair, 0) / self.total_exposure) * 100 class CorrelationEngine: """Rolling Pearson correlation matrix (Task 3.2). Replaces the hardcoded CORRELATION_CLUSTERS with a dynamic calculation based on the past 30 days of Close prices. """ def __init__(self): self.correlation_cache: Dict[Tuple[str, str], float] = {} self.last_update = None self.price_series: Dict[str, List[float]] = {} def update_series(self, historical_closes: Dict[str, List[float]]): """Feed 30 days of Close prices for all 28 pairs.""" self.price_series = historical_closes self.correlation_cache.clear() self.last_update = time.time() def get_correlation(self, pair_a: str, pair_b: str) -> float: """Get Pearson r between two pairs.""" key = tuple(sorted([pair_a, pair_b])) if key in self.correlation_cache: return self.correlation_cache[key] series_a = self.price_series.get(pair_a, []) series_b = self.price_series.get(pair_b, []) r = pearson_correlation(series_a, series_b) self.correlation_cache[key] = r return r def get_top_correlated( self, target_pair: str, n: int = 3, min_r: float = 0.75 ) -> List[Tuple[str, float]]: """Get top N pairs most correlated (|r| >= min_r) with target.""" results = [] for pair in self.price_series: if pair == target_pair: continue r = self.get_correlation(target_pair, pair) if abs(r) >= min_r: results.append((pair, r)) results.sort(key=lambda x: abs(x[1]), reverse=True) return results[:n] class BasketHedging: """Dynamic basket hedging using Pearson correlation (Task 3.2). Instead of hardcoded correlation clusters, uses CorrelationEngine to select the top 3 pairs with |r| >= 0.75 to the target cross-pair. """ def __init__(self, correlation_engine: CorrelationEngine = None): self.basket_positions = [] self.correlation_engine = correlation_engine or CorrelationEngine() def set_correlation_engine(self, engine: CorrelationEngine): self.correlation_engine = engine def get_correlated_pairs(self, pair: str) -> List[str]: """Get dynamically correlated pairs for basket hedging.""" top = self.correlation_engine.get_top_correlated(pair, n=3, min_r=0.75) return [p for p, r in top] def create_basket_hedge( self, primary_pair: str, primary_size: float, confluence_strength: float, current_prices: Dict[str, float] = None ) -> List[Dict]: correlated = self.get_correlated_pairs(primary_pair) self.basket_positions = [] if not correlated: return self.basket_positions hedge_ratio = 0.3 hedge_size = primary_size * hedge_ratio / len(correlated) for cp in correlated: entry = (current_prices or {}).get(cp, 0) self.basket_positions.append({ 'pair': cp, 'size': hedge_size, 'entry_price': entry, 'type': 'BASKET_HEDGE', 'primary_pair': primary_pair, }) return self.basket_positions def get_total_basket_exposure(self) -> float: return sum(h['size'] for h in self.basket_positions) class RiskManagementSystem: """Complete risk management system with portfolio-level exit (Task 3.3). Features: - Position sizing with spread penalty (Task 2.1) - Dynamic basket hedging via Pearson correlation (Task 3.2) - Aggregate portfolio P&L monitoring with dynamic profit target (Task 3.3) - Portfolio-based exit: close ALL trades when basket P&L > target """ def __init__(self, account_balance: float = 10000.0): self.account_balance = account_balance self._lock = threading.Lock() self.sizer = PositionSizer(account_balance, risk_per_trade=0.01) self.hedger = GridHedging(grid_levels=config.GRID_LEVELS) self.portfolio = PortfolioExposure(max_portfolio_leverage=config.MAX_PORTFOLIO_LEVERAGE) self.correlation_engine = CorrelationEngine() self.basket = BasketHedging(self.correlation_engine) self.trades = [] self.current_prices: Dict[str, float] = {} self.current_order_books: Dict[str, Dict] = {} # Portfolio exit monitor (Task 3.3) self._exit_monitor_running = False self._exit_monitor_thread: Optional[threading.Thread] = None self._exit_callbacks: List[Callable] = [] def update_prices(self, prices: Dict[str, float]): self.current_prices.update(prices) def update_order_books(self, order_books: Dict[str, Dict]): self.current_order_books.update(order_books) def update_correlation_data(self, historical_closes: Dict[str, List[float]]): """Feed 30-day close prices for dynamic correlation (Task 3.2).""" self.correlation_engine.update_series(historical_closes) def execute_signal( self, pair: str, confluence_strength: float, entry_price: float, use_hedging: bool = True, use_basket: bool = True, order_book: Dict = None, ) -> Optional[Dict]: """Execute a confluence signal with full risk management. Uses actual bid/ask/spread from order book (Task 2.1) for position sizing if available. """ bid = (order_book or {}).get('bid', entry_price) ask = (order_book or {}).get('ask', entry_price) spread = (order_book or {}).get('spread') position_size = self.sizer.calculate_position_size( pair, confluence_strength, entry_price, stop_loss_pips=50, bid=bid, ask=ask, spread=spread ) if not self.portfolio.can_add_position(position_size, self.account_balance): return None trade = { 'pair': pair, 'entry_price': entry_price, 'entry_bid': bid, 'entry_ask': ask, 'position_size': position_size, 'confluence_strength': confluence_strength, 'status': 'OPEN', } if use_hedging and confluence_strength > 70: trade['grid_hedges'] = self.hedger.create_hedge_grid( pair, entry_price, position_size ) if use_basket and confluence_strength > 60: trade['basket_hedges'] = self.basket.create_basket_hedge( pair, position_size, confluence_strength, self.current_prices ) if config.DEBUG: n_hedges = len(trade.get('basket_hedges', [])) print(f"[Risk] Created {n_hedges} dynamic basket hedges for {pair}") with self._lock: self.portfolio.add_position(pair, position_size) self.trades.append(trade) return trade def calculate_basket_pnl(self) -> float: """Calculate aggregate P&L across ALL open positions and hedges.""" total = 0.0 for trade in self.trades: if trade['status'] != 'OPEN': continue pair = trade['pair'] entry = trade['entry_price'] current = self.current_prices.get(pair, entry) delta = current - entry total += delta * trade['position_size'] * 100000 for hedge in trade.get('grid_hedges', []): h_current = self.current_prices.get(hedge['pair'], hedge['price']) h_delta = h_current - hedge['price'] total += h_delta * hedge['size'] * 100000 for hedge in trade.get('basket_hedges', []): h_current = self.current_prices.get(hedge['pair'], hedge['entry_price']) h_delta = h_current - hedge['entry_price'] total += h_delta * hedge['size'] * 100000 return total def get_dynamic_exit_target(self) -> float: """Dynamic profit target based on trade confidence (Task 3.3). Higher confidence trades get a larger profit target. Base: +1% of account balance. """ if not self.trades: return self.account_balance * 0.01 avg_confidence = sum( t.get('confluence_strength', 50) for t in self.trades if t['status'] == 'OPEN' ) n_open = max(len([t for t in self.trades if t['status'] == 'OPEN']), 1) avg_confidence /= n_open base_target = self.account_balance * 0.01 confidence_mult = avg_confidence / 50.0 # 1.0x at 50%, 2.0x at 100% return base_target * confidence_mult def _monitor_exit_loop(self): """Background loop monitoring basket P&L every second (Task 3.3). When aggregate net P&L surpasses the dynamic target, fires close_all_trades() automatically. """ while self._exit_monitor_running: if not self.trades: time.sleep(1) continue total_pnl = self.calculate_basket_pnl() target = self.get_dynamic_exit_target() if total_pnl > target: if config.DEBUG: print(f"[Risk] Portfolio exit triggered: P&L={total_pnl:.2f} target={target:.2f}") for cb in self._exit_callbacks: try: cb(total_pnl) except Exception: pass break time.sleep(1) def start_exit_monitor(self): """Start the background portfolio exit monitor (Task 3.3).""" if self._exit_monitor_running: return self._exit_monitor_running = True self._exit_monitor_thread = threading.Thread( target=self._monitor_exit_loop, daemon=True ) self._exit_monitor_thread.start() if config.DEBUG: print("[Risk] Portfolio exit monitor started") def on_portfolio_exit(self, callback: Callable): """Register a callback for when the portfolio exit fires.""" self._exit_callbacks.append(callback) def stop_exit_monitor(self): self._exit_monitor_running = False def should_exit_portfolio(self) -> Tuple[bool, float]: """Check if aggregate portfolio P&L has hit the profit target.""" total_pnl = self.calculate_basket_pnl() if total_pnl > self.get_dynamic_exit_target(): return True, total_pnl return False, total_pnl def close_trade(self, pair: str, exit_price: float) -> Optional[Dict]: with self._lock: result = self.sizer.close_position(pair, exit_price) if result: self.portfolio.remove_position(pair) return result def close_all_trades(self, exit_prices: Dict[str, float]): """Close all open trades at given exit prices.""" results = [] with self._lock: for trade in list(self.trades): if trade['status'] == 'OPEN': price = exit_prices.get(trade['pair'], trade['entry_price']) result = self.sizer.close_position(trade['pair'], price) if result: self.portfolio.remove_position(trade['pair']) results.append(result) self.stop_exit_monitor() return results def get_portfolio_summary(self) -> Dict: return { 'total_positions': len(self.portfolio.positions), 'total_exposure': self.portfolio.total_exposure, 'leverage_ratio': self.portfolio.get_leverage_ratio(self.account_balance), 'open_trades': len([t for t in self.trades if t['status'] == 'OPEN']), 'basket_pnl': self.calculate_basket_pnl(), 'exit_target': self.get_dynamic_exit_target(), 'account_balance': self.account_balance, }