gavindiaz/NexQuant
1
0
Fork 0
mirror of https://github.com/NicolasBohn/NexQuant.git synced 2026-07-27 23:47:46 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: close log file handle, fix RiskMgmt equity double-count, remove bare except

Browse Source
This commit is contained in:
TPTBusiness
2026-05-03 00:17:02 +02:00
parent 4eeb724ac5
commit ca003cd0f2
3 changed files with 139 additions and 135 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rdagent/components/backtesting/risk_management.py
+18 -20
View File
@@ -1,21 +1,19 @@
"""
Predix Risk Management - Korrelation, Portfolio-Optimierung
"""
import numpy as np
import pandas as pd
from pathlib import Path
from typing import Dict, List, Optional
from datetime import datetime
import json
class CorrelationAnalyzer:
def __init__(self, lookback: int = 60):
self.lookback = lookback
def calculate_matrix(self, returns: pd.DataFrame) -> pd.DataFrame:
return returns.dropna().corr()
def find_uncorrelated(self, corr: pd.DataFrame, threshold: float = 0.3) -> List[str]:
def find_uncorrelated(self, corr: pd.DataFrame, threshold: float = 0.3) -> list[str]:
result = []
for f in corr.columns:
others = [x for x in corr.columns if x != f]
@@ -28,9 +26,9 @@ class PortfolioOptimizer:
try:
w = np.linalg.inv(cov.values) @ exp_ret.values
return w / np.sum(w)
except:
except np.linalg.LinAlgError:
return np.ones(len(exp_ret)) / len(exp_ret)
def risk_parity(self, cov: pd.DataFrame, max_iter: int = 100) -> np.ndarray:
n = cov.shape[0]
w = np.ones(n) / n
@@ -53,36 +51,36 @@ class AdvancedRiskManager:
self.max_dd = max_dd
self.corr_analyzer = CorrelationAnalyzer()
self.optimizer = PortfolioOptimizer()
def check_limits(self, weights: np.ndarray, vol: float, dd: float) -> Dict[str, bool]:
def check_limits(self, weights: np.ndarray, vol: float, dd: float) -> dict[str, bool]:
return {
'position_limit': np.max(np.abs(weights)) <= self.max_pos,
'leverage_limit': np.sum(np.abs(weights)) <= self.max_lev,
'drawdown_limit': abs(dd) <= self.max_dd,
"position_limit": np.max(np.abs(weights)) <= self.max_pos,
"leverage_limit": np.sum(np.abs(weights)) <= self.max_lev,
"drawdown_limit": abs(dd) <= self.max_dd,
}
if __name__ == "__main__":
print("=== Risk Test ===")
np.random.seed(42)
n, names = 252, ['Mom', 'MeanRev', 'Vol', 'Volu', 'ML']
n, names = 252, ["Mom", "MeanRev", "Vol", "Volu", "ML"]
ret = pd.DataFrame(np.random.randn(n, 5), columns=names)
corr = CorrelationAnalyzer().calculate_matrix(ret)
print("Korrelationsmatrix:")
print(corr.round(2))
opt = PortfolioOptimizer()
exp_ret = pd.Series([0.1, 0.08, 0.06, 0.07, 0.12], index=names)
cov = ret.cov() * 252
mv = opt.mean_variance(exp_ret, cov)
print("\nMean-Variance:")
for n, w in zip(names, mv): print(f" {n}: {w:.2%}")
rp = opt.risk_parity(cov)
print("\nRisk Parity:")
for n, w in zip(names, rp): print(f" {n}: {w:.2%}")
rm = AdvancedRiskManager()
checks = rm.check_limits(mv, 0.15, -0.08)
print(f"\nLimits OK: {all(checks.values())}")
rdagent/components/backtesting/vbt_backtest.py
+20 -23
View File
@@ -19,7 +19,7 @@ Design goals
"""
from __future__ import annotations
from typing import Any, Dict, Optional
from typing import Any
import numpy as np
import pandas as pd
@@ -69,7 +69,7 @@ def _cross_check_with_vbt(
txn_cost: float,
manual_total_return: float,
freq: str,
) -> Optional[float]:
) -> float | None:
"""Run a vectorbt simulation and return its total_return for comparison."""
if not VBT_AVAILABLE:
return None
@@ -95,9 +95,9 @@ def backtest_signal(
txn_cost_bps: float = DEFAULT_TXN_COST_BPS,
freq: str = "1min",
bars_per_year: int = DEFAULT_BARS_PER_YEAR,
forward_returns: Optional[pd.Series] = None,
forward_returns: pd.Series | None = None,
cross_check: bool = False,
) -> Dict[str, Any]:
) -> dict[str, Any]:
"""
Run a single-asset backtest from a position signal.
@@ -204,7 +204,7 @@ def backtest_signal(
calmar = ann_return_arith / abs(max_dd) if max_dd < 0 else 0.0
trade_pnl = _compute_trade_pnl(position, strategy_returns)
n_trades = int(len(trade_pnl))
n_trades = len(trade_pnl)
n_position_changes = int((position.diff().fillna(0) != 0).sum())
if n_trades > 0:
@@ -216,7 +216,7 @@ def backtest_signal(
win_rate = 0.0
profit_factor = 0.0
ic: Optional[float] = None
ic: float | None = None
if forward_returns is not None:
fwd = pd.to_numeric(forward_returns, errors="coerce")
common = signal.index.intersection(fwd.dropna().index)
@@ -227,7 +227,7 @@ def backtest_signal(
ic_val = float(s.corr(f))
ic = ic_val if np.isfinite(ic_val) else None
result: Dict[str, Any] = {
result: dict[str, Any] = {
"status": "success",
"sharpe": sharpe,
"sortino": sortino,
@@ -244,7 +244,7 @@ def backtest_signal(
"volatility": volatility,
"n_trades": n_trades,
"n_position_changes": n_position_changes,
"n_bars": int(len(strategy_returns)),
"n_bars": len(strategy_returns),
"n_months": float(n_months),
"signal_long": int((signal > 0).sum()),
"signal_short": int((signal < 0).sum()),
@@ -293,7 +293,7 @@ def _apply_ftmo_mask(
daily_breaches = 0
total_breached = False
total_breach_ts: Optional[pd.Timestamp] = None
total_breach_ts: pd.Timestamp | None = None
current_day = None
day_start_eq = FTMO_INITIAL_CAPITAL
@@ -308,11 +308,8 @@ def _apply_ftmo_mask(
pos_i = float(signal.at[ts]) * leverage
ret_i = float(bar_ret.get(ts, 0.0))
cost_i = abs(pos_i - pos_prev) * txn_cost
ret_net = pos_prev * ret_i - cost_i
equity = equity * (1.0 + ret_net / FTMO_INITIAL_CAPITAL * FTMO_INITIAL_CAPITAL / equity
if equity > 0 else 1.0)
# Simpler: track as fraction
equity += FTMO_INITIAL_CAPITAL * ret_net
ret_frac = pos_prev * ret_i - cost_i
equity *= 1.0 + ret_frac if equity > 0 else 1.0
pos_prev = pos_i
if total_breached:
@@ -399,7 +396,7 @@ def walk_forward_rolling(
is_years: int = WF_IS_YEARS,
oos_years: int = WF_OOS_YEARS,
step_years: int = WF_STEP_YEARS,
) -> Dict[str, Any]:
) -> dict[str, Any]:
"""
Rolling walk-forward validation: multiple IS/OOS windows shifted by ``step_years``.
@@ -433,7 +430,7 @@ def walk_forward_rolling(
yr += step_years
continue
window: Dict[str, Any] = {
window: dict[str, Any] = {
"is_start": str(is_start.date()),
"is_end": str(is_end.date()),
"oos_start": str(is_end.date()),
@@ -475,11 +472,11 @@ def backtest_signal_ftmo(
stop_pips: float = FTMO_STOP_PIPS,
max_leverage: float = FTMO_MAX_LEVERAGE,
bars_per_year: int = DEFAULT_BARS_PER_YEAR,
forward_returns: Optional[pd.Series] = None,
oos_start: Optional[str] = OOS_START_DEFAULT,
forward_returns: pd.Series | None = None,
oos_start: str | None = OOS_START_DEFAULT,
wf_rolling: bool = False,
mc_n_permutations: int = 0,
) -> Dict[str, Any]:
) -> dict[str, Any]:
"""
FTMO-compliant backtest of a strategy signal on EUR/USD.
@@ -547,7 +544,7 @@ def backtest_signal_ftmo(
is_mask = close.index < oos_ts
oos_mask = close.index >= oos_ts
def _split_bt(mask: "pd.Series[bool]", prefix: str) -> None:
def _split_bt(mask: pd.Series[bool], prefix: str) -> None:
if mask.sum() < 100:
return
close_s = close.loc[mask]
@@ -602,7 +599,7 @@ def backtest_from_forward_returns(
forward_returns: pd.Series,
txn_cost_bps: float = DEFAULT_TXN_COST_BPS,
bars_per_year: int = DEFAULT_BARS_PER_YEAR,
) -> Dict[str, Any]:
) -> dict[str, Any]:
"""
Backtest a factor using sign(factor) as signal against forward returns.
@@ -640,7 +637,7 @@ def backtest_from_forward_returns(
ic = ic_val if np.isfinite(ic_val) else 0.0
trade_pnl = _compute_trade_pnl(position, strategy_returns)
n_trades = int(len(trade_pnl))
n_trades = len(trade_pnl)
win_rate = float((trade_pnl > 0).mean()) if n_trades > 0 else 0.0
ann_return = float(strategy_returns.mean() * bars_per_year)
@@ -656,7 +653,7 @@ def backtest_from_forward_returns(
"win_rate": win_rate,
"n_trades": n_trades,
"ic": ic,
"n_bars": int(len(strategy_returns)),
"n_bars": len(strategy_returns),
"txn_cost_bps": txn_cost_bps,
"bars_per_year": bars_per_year,
}