"""Sector concentration and return-correlation measurement (issue #65).
`docs/ideas.md` item 7 named this "the most under-measured risk here":
theme-driven discovery structurally concentrates a book into a handful of
correlated macro bets wearing different tickers, and neither
`max_position_pct` nor `max_total_exposure_pct` notices that the names move
together. This module is a measurement only — it computes numbers for an
existing report to print. It does not gate an order, is never called from
`execution/` or a strategy, and imports nothing from `brokers/` or
`execution/` itself.
Two independent measurements, matching the issue's two asks:
- `sector_exposure` — held positions' market value, grouped by sector, as a
fraction of the book. A symbol whose sector lookup fails or is
unclassified (an ETF, mainly) is grouped under `UNKNOWN_SECTOR`, an
explicit and visible bucket added into the total the same way a real
sector is — never silently dropped, and never folded into a real sector as
if it were known to belong there.
- `compute_correlations` — pairwise Pearson correlation of daily returns
across held positions, computed from the existing bar cache (no new data
source). A symbol with too little history to build a return series (a
fresh listing, a data outage) is left out of every pair rather than
treated as uncorrelated with everything else — "not computed" is not the
same claim as "zero risk".
A third helper, `sector_distribution_by_count`, answers the "discovered"
half of the issue's first ask: a discovered symbol carries no market value
until it becomes a position, so weighting it by dollars is meaningless —
counting it once each is the only honest way to show whether the *candidate
pool itself* is concentrated before any of it is bought.
"""
from __future__ import annotations
import logging
import math
from collections.abc import Sequence
from dataclasses import dataclass
from datetime import date, datetime, timedelta
from decimal import ROUND_HALF_UP, Decimal
from trader.domain import Bar, Position
from trader.errors import MarketDataError
from trader.marketdata.cache import BarCache
from trader.marketdata.sector import SectorProvider
__all__ = [
"DEFAULT_CORRELATION_LOOKBACK_DAYS",
"UNKNOWN_SECTOR",
"PortfolioCorrelation",
"SectorCount",
"SectorExposure",
"SymbolCorrelation",
"compute_correlations",
"sector_distribution_by_count",
"sector_exposure",
]
_logger = logging.getLogger("trader.reporting")
#: The explicit bucket for a symbol whose sector could not be determined —
#: no coverage (an ETF), a lookup failure, or an unclassified new listing.
#: Never blank, and never folded into a real sector.
UNKNOWN_SECTOR = "Unknown"
_PCT_QUANTUM = Decimal("0.01")
#: Default lookback for the correlation window. 90 calendar days is roughly
#: 60 trading sessions — enough for a Pearson correlation to mean something
#: without reaching so far back that a recently discovered, thinly-traded
#: name has no history at all inside the window.
DEFAULT_CORRELATION_LOOKBACK_DAYS = 90
#: Fewer daily returns than this and a pair's correlation is not computed at
#: all — a two- or three-point "correlation" is noise dressed as a number.
_MIN_RETURNS_FOR_CORRELATION = 10
#: Correlation is pairwise by definition; fewer than two symbols with
#: positions means there is no pair to even attempt.
_MIN_SYMBOLS_FOR_CORRELATION = 2
[docs]
@dataclass(frozen=True, slots=True)
class SectorExposure:
"""One sector's (or `UNKNOWN_SECTOR`'s) share of held market value."""
sector: str
market_value: Decimal
pct_of_total: Decimal
symbols: tuple[str, ...]
[docs]
def sector_exposure(
positions: Sequence[Position], sector_provider: SectorProvider
) -> list[SectorExposure]:
"""Held positions' market value, grouped by sector, richest first.
Weighted by `market_value` — this answers "how much of the book sits in
one sector", not "how many names". A sector lookup failure for one
symbol costs only that symbol (same isolation discipline as discovery
and the analyst-consensus fetch): it is grouped under `UNKNOWN_SECTOR`
and logged, never allowed to abort the rest of the book's grouping.
"""
if not positions:
return []
totals: dict[str, Decimal] = {}
symbols_by_sector: dict[str, list[str]] = {}
grand_total = Decimal(0)
for position in positions:
sector = _lookup_sector(position.symbol, sector_provider)
totals[sector] = totals.get(sector, Decimal(0)) + position.market_value
symbols_by_sector.setdefault(sector, []).append(position.symbol)
grand_total += position.market_value
if grand_total == 0:
# Every position marked at zero market value is not a real book to
# report a percentage breakdown of — an empty result, not a
# fabricated 0.00% split. Distinct from `positions` being empty,
# handled above.
return []
results = [
SectorExposure(
sector=sector,
market_value=value,
pct_of_total=(value / grand_total * 100).quantize(
_PCT_QUANTUM, rounding=ROUND_HALF_UP
),
symbols=tuple(symbols_by_sector[sector]),
)
for sector, value in totals.items()
]
results.sort(key=lambda row: row.market_value, reverse=True)
return results
def _lookup_sector(symbol: str, sector_provider: SectorProvider) -> str:
try:
info = sector_provider.get_sector_info(symbol)
except MarketDataError:
_logger.warning("sector lookup failed for %s", symbol, exc_info=True)
return UNKNOWN_SECTOR
return info.sector or UNKNOWN_SECTOR
[docs]
@dataclass(frozen=True, slots=True)
class SectorCount:
"""One sector's (or `UNKNOWN_SECTOR`'s) share of a symbol list, by count."""
sector: str
count: int
pct_of_total: Decimal
symbols: tuple[str, ...]
[docs]
def sector_distribution_by_count(
symbols: Sequence[str], sector_provider: SectorProvider
) -> list[SectorCount]:
"""A symbol list's sector breakdown, weighted one-per-symbol.
For a set of symbols that carry no market value yet — discovered
candidates that are not (or not all) held positions — counting is the
only weighting that makes sense; see the module docstring. Duplicate
symbols in `symbols` are deduplicated first, so a candidate discovered
on more than one theme is not counted twice.
"""
unique = sorted(set(symbols))
if not unique:
return []
counts: dict[str, int] = {}
symbols_by_sector: dict[str, list[str]] = {}
for symbol in unique:
sector = _lookup_sector(symbol, sector_provider)
counts[sector] = counts.get(sector, 0) + 1
symbols_by_sector.setdefault(sector, []).append(symbol)
total = len(unique)
results = [
SectorCount(
sector=sector,
count=count,
pct_of_total=(Decimal(count) / total * 100).quantize(
_PCT_QUANTUM, rounding=ROUND_HALF_UP
),
symbols=tuple(symbols_by_sector[sector]),
)
for sector, count in counts.items()
]
results.sort(key=lambda row: row.count, reverse=True)
return results
[docs]
@dataclass(frozen=True, slots=True)
class SymbolCorrelation:
"""Pearson correlation of daily returns between two held symbols."""
symbol_a: str
symbol_b: str
correlation: float
[docs]
@dataclass(frozen=True, slots=True)
class PortfolioCorrelation:
"""Every computable pair, and their average.
`average_pairwise` is `None` — never `0.0` — when fewer than two symbols
had enough return history to compute even one pair: "no correlation
data" and "measured, and it's zero" are different claims, and this
module never lets the first read as the second.
"""
pairwise: tuple[SymbolCorrelation, ...]
average_pairwise: float | None
[docs]
def compute_correlations(
positions: Sequence[Position],
bar_cache: BarCache,
*,
now: datetime,
lookback_days: int = DEFAULT_CORRELATION_LOOKBACK_DAYS,
) -> PortfolioCorrelation:
"""Pairwise correlation across held positions' daily returns.
Bars come from the existing `BarCache` — no new data source, per the
issue's own constraint. A symbol whose bar fetch fails, or that has too
few bars in the window (`_MIN_RETURNS_FOR_CORRELATION`), is left out of
every pair rather than defaulting to any particular correlation value.
"""
symbols = sorted({position.symbol for position in positions})
if len(symbols) < _MIN_SYMBOLS_FOR_CORRELATION:
return PortfolioCorrelation(pairwise=(), average_pairwise=None)
start = now - timedelta(days=lookback_days)
returns_by_symbol: dict[str, dict[date, float]] = {}
for symbol in symbols:
try:
bars = bar_cache.ensure(symbol, start, now, interval="1d")
except MarketDataError:
_logger.warning(
"bar fetch failed for %s while computing correlation",
symbol,
exc_info=True,
)
continue
series = _daily_returns(bars)
if len(series) >= _MIN_RETURNS_FOR_CORRELATION:
returns_by_symbol[symbol] = series
pairs: list[SymbolCorrelation] = []
usable = sorted(returns_by_symbol)
for i, symbol_a in enumerate(usable):
for symbol_b in usable[i + 1 :]:
correlation = _pearson(
returns_by_symbol[symbol_a], returns_by_symbol[symbol_b]
)
if correlation is not None:
pairs.append(SymbolCorrelation(symbol_a, symbol_b, correlation))
average = sum(pair.correlation for pair in pairs) / len(pairs) if pairs else None
return PortfolioCorrelation(pairwise=tuple(pairs), average_pairwise=average)
def _daily_returns(bars: Sequence[Bar]) -> dict[date, float]:
"""Close-to-close percentage returns, keyed by the later bar's date.
A `dict` keyed by date, not a plain list, because two symbols' bars are
not guaranteed to share every trading day (a new listing, a fetch that
stopped short) — `_pearson` aligns two series on their *shared* dates
rather than assuming position `i` in one list means the same day as
position `i` in the other.
"""
ordered = sorted(bars, key=lambda bar: bar.timestamp)
returns: dict[date, float] = {}
previous_close: Decimal | None = None
for bar in ordered:
if previous_close is not None and previous_close != 0:
returns[bar.timestamp.date()] = float(
(bar.close - previous_close) / previous_close
)
previous_close = bar.close
return returns
def _pearson(series_a: dict[date, float], series_b: dict[date, float]) -> float | None:
"""Pearson correlation over the dates both series share.
`None` when fewer than `_MIN_RETURNS_FOR_CORRELATION` shared dates
remain after aligning, or when either series has zero variance (a flat
return series makes the coefficient undefined, not zero — dividing by a
zero standard deviation would otherwise raise or silently produce NaN).
"""
shared = sorted(set(series_a) & set(series_b))
if len(shared) < _MIN_RETURNS_FOR_CORRELATION:
return None
xs = [series_a[d] for d in shared]
ys = [series_b[d] for d in shared]
mean_x = sum(xs) / len(xs)
mean_y = sum(ys) / len(ys)
covariance = sum((x - mean_x) * (y - mean_y) for x, y in zip(xs, ys, strict=True))
variance_x = sum((x - mean_x) ** 2 for x in xs)
variance_y = sum((y - mean_y) ** 2 for y in ys)
denominator = math.sqrt(variance_x * variance_y)
if denominator == 0:
return None
return covariance / denominator