"""Base class for alphas. An alpha maps a wide close matrix (date index × symbol_id columns) to signed position weights. Subclasses implement :meth:`signal` — the raw, unnormalized score. The base class turns a signal into cross-sectionally z-scored weights via :meth:`to_weights` (override it for a different normalization). """ from abc import ABC, abstractmethod import numpy as np import pandas as pd class BaseAlpha(ABC): """A position-weight alpha over a cross-section of stocks. Concrete subclasses must set a unique class-level :attr:`name` (the registry key) and implement :meth:`signal`. Construct subclasses with their own typed parameters (e.g. ``lookback``); the factory passes only the parameters a given ``__init__`` accepts. """ #: Unique registry key. Every concrete alpha must set this to a non-empty str. name: str = "" @abstractmethod def signal(self, close: pd.DataFrame) -> pd.DataFrame: """Compute the raw signal. Args: close: Wide close prices, date index × ``symbol_id`` columns. Returns: A wide DataFrame aligned to ``close`` where higher values indicate a stronger long. Use NaN where the signal is undefined. """ def to_weights(self, signal: pd.DataFrame) -> pd.DataFrame: """Cross-sectionally z-score a signal into signed position weights. Each date is demeaned and scaled by its cross-sectional std; undefined cells become a 0 weight. Override for a custom scheme (rank, neutralized, capped, etc.). """ signal = signal.dropna(how="all") demeaned = signal.subtract(signal.mean(axis=1), axis=0) std = signal.std(axis=1).replace(0, np.nan) weights = demeaned.divide(std, axis=0) return weights.fillna(0.0) def weights(self, close: pd.DataFrame) -> pd.DataFrame: """Full pipeline for one alpha: raw signal → normalized weights.""" return self.to_weights(self.signal(close)) def __repr__(self) -> str: params = ", ".join(f"{k}={v!r}" for k, v in vars(self).items()) return f"{type(self).__name__}({params})"