Yuxuan Yan
91 lines
3.3 KiB
Python
91 lines
3.3 KiB
Python
"""Layer-1 (research) evaluation of a portfolio's target weights.
|
|
|
|
This is the WorldQuant-style research view: continuous target weights, no lot or
|
|
trading constraints. Metrics are return / Sharpe / turnover / max-drawdown /
|
|
**Fitness**. There is deliberately **no IC/IR** — consistent with the repo's
|
|
convention that an alpha is a position weight, not a return predictor.
|
|
|
|
Return convention (documented): the target weight formed from information at
|
|
date ``t`` earns the *next* period's close-to-close return, i.e. weights are
|
|
shifted one day relative to realized returns, so there is no look-ahead:
|
|
``R_t = sum_i w_{i,t} · r_{i,t+1}`` normalized by gross exposure.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
|
|
#: WorldQuant fitness floor on turnover (avoids dividing by ~0 turnover).
|
|
_TURNOVER_FLOOR = 0.125
|
|
|
|
|
|
def evaluate_portfolio(positions_df: pd.DataFrame, data_df: pd.DataFrame) -> dict:
|
|
"""Evaluate target weights as a continuous research portfolio.
|
|
|
|
Args:
|
|
positions_df: POSITION_COLUMNS (uses ``target_weight``; zero-gross
|
|
construction carry dates remain flat in this research view).
|
|
data_df: DATA_COLUMNS (uses ``close`` for returns).
|
|
|
|
Returns:
|
|
Dict with ``cumulative_return, sharpe_annual, turnover_annual,
|
|
max_drawdown, fitness, hit_rate, n_dates``. No IC key.
|
|
"""
|
|
close = data_df.pivot_table(
|
|
index="date", columns="symbol_id", values="close", aggfunc="first"
|
|
).sort_index()
|
|
returns = close.pct_change()
|
|
|
|
weights = positions_df.pivot_table(
|
|
index="date", columns="symbol_id", values="target_weight", aggfunc="first"
|
|
).sort_index()
|
|
|
|
common = weights.index.intersection(returns.index)
|
|
weights = weights.loc[common]
|
|
returns = returns.loc[common]
|
|
|
|
empty = {
|
|
"cumulative_return": 0.0, "sharpe_annual": 0.0, "turnover_annual": 0.0,
|
|
"max_drawdown": 0.0, "fitness": 0.0, "hit_rate": 0.0,
|
|
"n_dates": len(common),
|
|
}
|
|
if len(common) < 3:
|
|
return empty
|
|
|
|
gross = weights.abs().sum(axis=1)
|
|
# Weights at t earn the return from t to t+1: shift returns back by one.
|
|
fwd = returns.shift(-1)
|
|
daily = (weights * fwd).sum(axis=1) / gross.replace(0.0, np.nan)
|
|
daily = daily.dropna()
|
|
if len(daily) < 2:
|
|
return empty
|
|
|
|
cumulative_return = float((1.0 + daily).prod() - 1.0)
|
|
mu, sigma = daily.mean(), daily.std()
|
|
sharpe_annual = float(np.sqrt(252) * mu / sigma) if sigma > 0 else 0.0
|
|
|
|
weight_change = weights.diff().abs().sum(axis=1)
|
|
prev_gross = gross.shift(1)
|
|
daily_turnover = (weight_change / prev_gross.replace(0.0, np.nan)).dropna()
|
|
turnover_annual = float(daily_turnover.mean() * 252)
|
|
|
|
equity = (1.0 + daily).cumprod()
|
|
drawdown = (equity - equity.cummax()) / equity.cummax()
|
|
max_drawdown = float(drawdown.min())
|
|
|
|
# Fitness = Sharpe · sqrt(|annualized return| / max(annual turnover, floor)).
|
|
ann_return = float(mu * 252)
|
|
denom = max(turnover_annual, _TURNOVER_FLOOR)
|
|
fitness = float(sharpe_annual * np.sqrt(abs(ann_return) / denom)) if denom > 0 else 0.0
|
|
|
|
return {
|
|
"cumulative_return": cumulative_return,
|
|
"sharpe_annual": sharpe_annual,
|
|
"turnover_annual": turnover_annual,
|
|
"max_drawdown": max_drawdown,
|
|
"fitness": fitness,
|
|
"hit_rate": float((daily > 0).mean()),
|
|
"n_dates": int(len(daily)),
|
|
}
|