dam/Trading-Bot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Status when starting a new project on Chat GPT

Browse Source
This commit is contained in:
DaM
2026-03-01 16:50:15 +01:00
parent 547a909965
commit 2ec51d0daa
10 changed files with 1232 additions and 510 deletions
Download Patch File Download Diff File Expand all files Collapse all files

170
src/calibration/strategies_inspector.py
View File

@@ -21,27 +21,11 @@ from src.risk.sizing.percent_risk import PercentRiskSizer
# --------------------------------------------------
# Strategy registry (con metadata de parámetros)
# --------------------------------------------------
from src.strategies.registry import STRATEGY_REGISTRY
from src.strategies.moving_average import MovingAverageCrossover
from src.strategies.rsi_strategy import RSIStrategy
from src.strategies.buy_and_hold import BuyAndHold
STRATEGY_REGISTRY = {
"moving_average": {
"class": MovingAverageCrossover,
"params": ["fast_period", "slow_period"],
},
"rsi": {
"class": RSIStrategy,
"params": ["rsi_period", "overbought", "oversold"],
},
"buy_and_hold": {
"class": BuyAndHold,
"params": [],
},
}
# --------------------------------------------------
# Helpers
# --------------------------------------------------
@@ -49,15 +33,24 @@ STRATEGY_REGISTRY = {
def list_available_strategies() -> List[Dict[str, Any]]:
"""
Devuelve metadata completa para UI.
Usa parameters_schema() como fuente de verdad.
"""
out = []
for sid, entry in STRATEGY_REGISTRY.items():
out: List[Dict[str, Any]] = []
for strategy_id, strategy_class in STRATEGY_REGISTRY.items():
if not hasattr(strategy_class, "parameters_schema"):
continue
schema = strategy_class.parameters_schema()
out.append({
"strategy_id": sid,
"name": entry["class"].__name__,
"params": entry["params"],
"tags": [], # puedes rellenar más adelante
"strategy_id": strategy_id,
"name": strategy_class.__name__,
"params": list(schema.keys()),
"parameters_schema": schema, # 🔥 ahora enviamos schema completo
"tags": [],
})
return out
@@ -67,7 +60,7 @@ def _build_stop_loss(stop_schema) -> object | None:
if stop_schema.type == "fixed":
return FixedStop(stop_fraction=float(stop_schema.stop_fraction))
if stop_schema.type == "trailing":
return TrailingStop(stop_fraction=float(stop_schema.stop_fraction))
return TrailingStop(trailing_fraction=float(stop_schema.stop_fraction))
if stop_schema.type == "atr":
return ATRStop(
atr_period=int(stop_schema.atr_period),
@@ -94,27 +87,6 @@ def _accumulate_equity(initial: float, returns_pct: List[float]) -> List[float]:
return eq
def _build_param_values(min_v: float, max_v: float, step: float) -> List[float]:
min_v = float(min_v)
max_v = float(max_v)
step = float(step)
# Valor único si min == max
if min_v == max_v:
return [min_v]
# Valor único si step <= 1
if step <= 1:
return [min_v]
values = []
v = min_v
while v <= max_v:
values.append(v)
v += step
return values
# --------------------------------------------------
# Main
# --------------------------------------------------
@@ -162,14 +134,21 @@ def inspect_strategies_config(
step_td = pd.Timedelta(days=int(payload.wf.step_days or payload.wf.test_days))
overall_status = "ok"
log.info(f"🔥 Strategies received: {len(payload.strategies)}")
results: List[Dict[str, Any]] = []
series: Dict[str, Any] = {"strategies": {}} if include_series else {}
log.info(f"🔥 Strategies received: {len(payload.strategies)}")
for sel in payload.strategies:
sid = sel.strategy_id
entry = STRATEGY_REGISTRY.get(sid)
log.info(f"🧠 Step3 | Processing strategy: {sid}")
if entry is None:
results.append({
"strategy_id": sid,
@@ -183,10 +162,13 @@ def inspect_strategies_config(
"windows": [],
})
overall_status = "fail"
log.error(f"❌ Strategy not found in registry: {sid}")
continue
strategy_class = entry["class"]
valid_params = set(entry["params"])
strategy_class = STRATEGY_REGISTRY[sid]
schema = strategy_class.parameters_schema()
valid_params = set(schema.keys())
range_params = set(sel.parameters.keys())
@@ -207,19 +189,15 @@ def inspect_strategies_config(
})
overall_status = "fail"
continue
# --------------------------------------------------
# Convert ranges -> param_grid real
# Build fixed_params (VALIDATION MODE)
# --------------------------------------------------
param_grid = {}
fixed_params = {}
for pname, pvalue in sel.parameters.items():
fixed_params[pname] = pvalue
for pname, prange in sel.parameters.items():
values = _build_param_values(
min_v=prange.min,
max_v=prange.max,
step=prange.step,
)
param_grid[pname] = values
# Wrapper sizer
class _CappedSizer(type(base_sizer)):
@@ -248,7 +226,8 @@ def inspect_strategies_config(
try:
wf = WalkForwardValidator(
strategy_class=strategy_class,
param_grid=param_grid,
param_grid=None,
fixed_params=fixed_params,
data=df,
train_window=train_td,
test_window=test_td,
@@ -256,49 +235,60 @@ def inspect_strategies_config(
initial_capital=float(payload.account_equity),
commission=float(payload.commission),
slippage=float(payload.slippage),
optimizer_metric=str(payload.optimization.optimizer_metric),
position_sizer=capped_sizer,
stop_loss=stop_loss,
max_combinations=int(payload.optimization.max_combinations),
progress_callback=progress_callback,
)
wf_res = wf.run()
win_df: pd.DataFrame = wf_res["windows"]
oos_returns = []
oos_dd = []
warnings_list = []
n_windows = 0
if win_df is None or win_df.empty:
status = "fail"
msg = "WF produced no valid windows"
overall_status = "fail"
windows_out = []
oos_returns = []
status = "warning"
msg = "No closed trades in OOS"
warnings_list.append("Walk-forward produced no closed trades.")
else:
trades = win_df["trades"].astype(int).tolist()
too_few = sum(t < int(payload.optimization.min_trades_test) for t in trades)
oos_returns = win_df["return_pct"].tolist()
oos_dd = win_df["max_dd_pct"].tolist()
n_windows = len(win_df)
if too_few > 0:
status = "warning"
msg = f"{too_few} windows below min_trades_test"
if overall_status == "ok":
overall_status = "warning"
else:
status = "ok"
msg = "WF OK"
trades = win_df["trades"].astype(int).tolist()
too_few = sum(t < int(payload.wf.min_trades_test) for t in trades)
windows_out = []
for _, r in win_df.iterrows():
windows_out.append({
"window": int(r["window"]),
"train_start": str(r["train_start"]),
"train_end": str(r["train_end"]),
"test_start": str(r["test_start"]),
"test_end": str(r["test_end"]),
"return_pct": float(r["return_pct"]),
"sharpe": float(r["sharpe"]),
"max_dd_pct": float(r["max_dd_pct"]),
"trades": int(r["trades"]),
"params": dict(r["params"]) if isinstance(r["params"], dict) else r["params"],
})
if too_few > 0:
warnings_list.append(
f"{too_few} test windows have fewer than {payload.wf.min_trades_test} trades"
)
windows_out = []
if warnings_list:
status = "warning"
msg = "Validation completed with warnings"
if overall_status == "ok":
overall_status = "warning"
else:
status = "ok"
msg = "WF OK"
for _, r in win_df.iterrows():
windows_out.append({
"window": int(r["window"]),
"train_start": str(r["train_start"]),
"train_end": str(r["train_end"]),
"test_start": str(r["test_start"]),
"test_end": str(r["test_end"]),
"return_pct": float(r["return_pct"]),
"sharpe": float(r["sharpe"]),
"max_dd_pct": float(r["max_dd_pct"]),
"trades": int(r["trades"]),
"params": dict(r["params"]) if isinstance(r["params"], dict) else r["params"],
})
oos_returns = win_df["return_pct"].astype(float).tolist()
@@ -311,6 +301,7 @@ def inspect_strategies_config(
"strategy_id": sid,
"status": status,
"message": msg,
"warnings": warnings_list if status == "warning" else [],
"n_windows": int(len(windows_out)),
"oos_final_equity": oos_final,
"oos_total_return_pct": float(oos_total_return),
@@ -323,6 +314,7 @@ def inspect_strategies_config(
series["strategies"][sid] = {
"window_returns_pct": oos_returns,
"window_equity": eq_curve,
"window_trades": win_df["trades"].tolist(),
}
except Exception as e: