Yuxuan Yan
72 lines
2.6 KiB
Python
72 lines
2.6 KiB
Python
"""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
|
||
|
||
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 either :meth:`signal` or :meth:`signal_from_data`.
|
||
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 = ""
|
||
|
||
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.
|
||
"""
|
||
raise NotImplementedError(
|
||
f"{type(self).__name__} must implement signal() or signal_from_data()"
|
||
)
|
||
|
||
def signal_from_data(
|
||
self,
|
||
data: pd.DataFrame,
|
||
close: pd.DataFrame,
|
||
) -> pd.DataFrame:
|
||
"""Compute the raw signal from long daily data plus wide closes.
|
||
|
||
Feature-aware alphas can override this to pivot joined feature columns
|
||
from ``data``. The default preserves the existing close-only alpha API.
|
||
"""
|
||
return self.signal(close)
|
||
|
||
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})"
|