Celes Renata
4ffde8cc06
- Database migration 018 with 13 tables for trading engine state - Trading engine service (services/trading/) with 12 pure computation modules: position sizer, stop-loss manager, reserve pool, circuit breaker, risk tier controller, correlation matrix, tax lots, trading window, gradual entry, notifications, micro-trading, backtester - Core TradingEngine with pre-trade evaluation pipeline and integration wiring - FastAPI HTTP service with 14 endpoints (health, config, decisions, metrics, backtest) - Performance tracker with Sharpe ratio, drawdown, profit factor computation - 194 Python tests (165 property-based + 29 integration) - Frontend: 13 TanStack Query hooks, 7 dashboard panels, tabbed Trading Engine page - Helm chart entry, network policy, nginx proxy, ingress for trading-engine - Shared infrastructure: enums, Redis keys, TradingConfig in AppConfig
68 lines
2.0 KiB
Python
68 lines
2.0 KiB
Python
"""Correlation matrix operations for portfolio diversification.
|
|
|
|
Feature: autonomous-trading-engine
|
|
|
|
Pure computation module for price correlation coefficients between
|
|
tracked companies. Used by the Position Sizer to prevent
|
|
over-concentration in correlated positions.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from services.trading.models import OpenPosition
|
|
|
|
|
|
class CorrelationMatrix:
|
|
"""In-memory correlation matrix for ticker pairs.
|
|
|
|
Stores pairwise correlation coefficients and provides lookup
|
|
methods for individual pairs and weighted portfolio averages.
|
|
"""
|
|
|
|
def __init__(self) -> None:
|
|
self._data: dict[tuple[str, str], float] = {}
|
|
|
|
def load(self, data: dict[tuple[str, str], float]) -> None:
|
|
"""Load correlation data from a dict of (ticker_a, ticker_b) -> coefficient."""
|
|
self._data = dict(data)
|
|
|
|
def get_correlation(self, ticker_a: str, ticker_b: str) -> float:
|
|
"""Return the correlation coefficient for a ticker pair.
|
|
|
|
Checks both orderings (a, b) and (b, a). Returns 0.0 if the
|
|
pair is not in the matrix.
|
|
"""
|
|
if ticker_a == ticker_b:
|
|
return 1.0
|
|
return self._data.get(
|
|
(ticker_a, ticker_b),
|
|
self._data.get((ticker_b, ticker_a), 0.0),
|
|
)
|
|
|
|
def get_portfolio_correlation(
|
|
self,
|
|
candidate: str,
|
|
positions: list[OpenPosition],
|
|
) -> float:
|
|
"""Compute weighted average correlation between candidate and positions.
|
|
|
|
Weights are based on each position's market_value. Returns 0.0
|
|
if there are no positions or total weight is zero.
|
|
"""
|
|
if not positions:
|
|
return 0.0
|
|
|
|
total_weight = 0.0
|
|
weighted_corr = 0.0
|
|
|
|
for pos in positions:
|
|
corr = self.get_correlation(candidate, pos.ticker)
|
|
weight = pos.market_value
|
|
weighted_corr += corr * weight
|
|
total_weight += weight
|
|
|
|
if total_weight == 0.0:
|
|
return 0.0
|
|
|
|
return weighted_corr / total_weight
|