fcostalonga/Trading
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

aggiunti salvataggi plot e contatore avanzamento

Browse Source
This commit is contained in:
fredmaloggia
2025-12-30 15:07:30 +01:00
parent 5f668cbb6d
commit e8f57afa99
3 changed files with 405 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

130
Trading Pattern Recon w Hurst - Stocks USA.py
View File

@@ -23,6 +23,7 @@ import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import requests
import time
# ------------------------------------------------------------
# shared_utils import (local file next to this script)
@@ -515,6 +516,39 @@ def plot_heatmap_monthly(r: pd.Series, title: str):
plt.tight_layout()
return fig
# ------------------------------------------------------------
# Progress timer (post-test checkpoints)
# ------------------------------------------------------------
def _format_eta(seconds):
if seconds is None or seconds != seconds:
return 'n/a'
seconds = max(0, int(round(seconds)))
minutes, secs = divmod(seconds, 60)
hours, minutes = divmod(minutes, 60)
if hours:
return f"{hours}h {minutes:02d}m {secs:02d}s"
return f"{minutes}m {secs:02d}s"
_post_timer = {'t0': None, 'tprev': None, 'total': None, 'done': 0}
def start_post_timer(total_steps: int):
_post_timer['t0'] = time.perf_counter()
_post_timer['tprev'] = _post_timer['t0']
_post_timer['total'] = total_steps
_post_timer['done'] = 0
def checkpoint_post_timer(label: str):
if _post_timer['t0'] is None or _post_timer['total'] is None:
return
_post_timer['done'] += 1
now = time.perf_counter()
step_dt = now - _post_timer['tprev']
total_dt = now - _post_timer['t0']
avg = total_dt / max(_post_timer['done'], 1)
eta = avg * max(_post_timer['total'] - _post_timer['done'], 0)
print(f"[TIMER] post {_post_timer['done']}/{_post_timer['total']} {label} - step {step_dt:.2f}s, total {total_dt:.2f}s, ETA {_format_eta(eta)}")
_post_timer['tprev'] = now
def _portfolio_metric_row(name: str, r: pd.Series) -> dict:
r = pd.to_numeric(r, errors="coerce").fillna(0.0)
@@ -606,6 +640,76 @@ def make_active_weights(
return res
def plot_portfolio_composition_fixed(
weights: pd.DataFrame,
title: str,
save_path: Path | None = None,
max_legend: int = 20,
):
"""Stacked area dei pesi nel tempo (pesi attivi + Cash)."""
if weights is None or getattr(weights, "empty", True):
print(f"[SKIP] Nessun peso per: {title}")
return
W = weights.copy().apply(pd.to_numeric, errors="coerce").fillna(0.0)
if W.index.has_duplicates:
W = W[~W.index.duplicated(keep="last")]
W = W.sort_index()
keep_cols = [c for c in W.columns if float(np.abs(W[c]).sum()) > 0.0]
if not keep_cols:
print(f"[SKIP] Tutti i pesi sono zero per: {title}")
return
W = W[keep_cols]
if len(W.index) < 2:
print(f"[SKIP] Serie troppo corta per: {title}")
return
avg_w = W.mean(0).sort_values(ascending=False)
ordered = avg_w.index.tolist()
if "Cash" in ordered:
ordered = [c for c in ordered if c != "Cash"] + ["Cash"]
if len(ordered) > max_legend:
head = ordered[:max_legend]
if "Cash" not in head and "Cash" in ordered:
head = head[:-1] + ["Cash"]
tail = [c for c in ordered if c not in head]
W_show = W[head].copy()
if tail:
W_show["Other"] = W[tail].sum(1)
ordered = head + ["Other"]
else:
ordered = head
else:
W_show = W[ordered].copy()
cmap = plt.colormaps.get_cmap("tab20")
colors = [cmap(i % cmap.N) for i in range(len(ordered))]
fig, ax = plt.subplots(figsize=(11, 6))
ax.stackplot(W_show.index, [W_show[c].values for c in ordered], labels=ordered, colors=colors)
ax.set_title(f"Composizione portafoglio nel tempo - {title}")
ymax = float(np.nanmax(W_show.sum(1).values))
if not np.isfinite(ymax) or ymax <= 0:
ymax = 1.0
ax.set_ylim(0, max(1.0, ymax))
ax.grid(True, alpha=0.3)
ax.set_ylabel("Peso")
ax.set_yticks(ax.get_yticks())
ax.set_yticklabels([f"{y*100:.0f}%" for y in ax.get_yticks()])
ncol = 2 if len(ordered) > 10 else 1
ax.legend(loc="upper left", bbox_to_anchor=(1.01, 1), frameon=False, ncol=ncol, title="Ticker")
fig.tight_layout()
if save_path:
fig.savefig(save_path, dpi=150, bbox_inches="tight")
print(f"[INFO] Salvato: {save_path}")
plt.close(fig)
def _build_dynamic_portfolio_returns(
wide_pnl: pd.DataFrame,
wide_sig: pd.DataFrame,
@@ -823,6 +927,7 @@ def calibrate_score_weights(
# MAIN
# ------------------------------------------------------------
def main():
start_post_timer(5)
# 1) Fetch prices
prices = {}
for tkr in TICKERS:
@@ -833,14 +938,19 @@ def main():
print(f"[WARN] Skip {tkr}: {e}")
if len(prices) < 5:
raise RuntimeError(f"Pochi ticker validi ({len(prices)}). Controlla TICKERS e/o endpoint.")
checkpoint_post_timer("Price fetch")
# 2) Backtest each ticker
hurst_rows = []
summary_rows = []
signals_rows = []
for tkr, px in prices.items():
total = len(prices)
for i, (tkr, px) in enumerate(prices.items(), 1):
print(f"[{i}/{total}] Testing {tkr} ...")
if not isinstance(px, pd.DataFrame) or "AdjClose" not in px.columns:
print(f"[WARN] Serie senza AdjClose per {tkr}: skip")
continue
@@ -892,6 +1002,8 @@ def main():
})
summary_rows.append(stats)
checkpoint_post_timer("Per-ticker backtest")
if not signals_rows:
raise RuntimeError("Nessun ticker backtestato con successo.")
@@ -958,6 +1070,8 @@ def main():
eq_eq = equity_from_returns(ret_eq).rename("Eq_EqW_TopN")
eq_rp = equity_from_returns(ret_rp).rename("Eq_RP_TopN")
checkpoint_post_timer("Portfolio build")
# 5) Plots
plt.figure(figsize=(10, 5))
plt.plot(eq_eq, label=f"Equal Weight (Top{TOP_N})")
@@ -977,6 +1091,19 @@ def main():
plt.savefig(PLOT_DIR / "heatmap_rp.png", dpi=150)
plt.show()
plot_portfolio_composition_fixed(
dyn_port["w_eq_act"],
"Equal Weight (active + Cash)",
PLOT_DIR / "composition_equal_weight_active.png",
)
plot_portfolio_composition_fixed(
dyn_port["w_rp_act"],
"Risk Parity (active + Cash)",
PLOT_DIR / "composition_risk_parity_active.png",
)
checkpoint_post_timer("Plots")
# 6) Save outputs
hurst_df.to_csv(OUT_DIR / "hurst.csv", index=False)
forward_bt_summary.to_csv(OUT_DIR / "forward_bt_summary.csv", index=False)
@@ -987,6 +1114,7 @@ def main():
pd.Series(base_tickers, name="TopN_Tickers").to_csv(OUT_DIR / "topn_tickers.csv", index=False)
save_portfolio_metrics(ret_eq, ret_rp, OUT_DIR / "portfolio_metrics.xlsx", TOP_N)
checkpoint_post_timer("Exports")
print(f"\nSaved to: {OUT_DIR.resolve()}")