1f45d77e4e
- generate_dashboard.py: adauga build_ferestre() (auto-scan edge x durata x fiabilitate, nimic hardcodat) + sheet date_grafic; scoate grila de ferestre pe formule din build_dashboard() - sterge scripts/generate_ferestre_v2.py si data/Ferestre_v2.xlsx (inlocuite) - generate_template.py: Dashboard pur-tabular (fara grila ferestre pe formule) - CLAUDE.md: documenteaza modelul combinat (un fisier Dashboard.xlsx) Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
769 lines
32 KiB
Python
769 lines
32 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""Generează data/Dashboard.xlsx dintr-un snapshot al data/backtest.xlsx.
|
||
|
||
UN SINGUR fișier de analiză. Citește backtest.xlsx (read-only, data_only=True)
|
||
și scrie data/Dashboard.xlsx cu TOATE analizele, în sheet-uri separate:
|
||
|
||
• Config — inputurile reale (lot, limite prop) + formule (din build_config).
|
||
• Trades — snapshot static ascuns (valori R_/$_ deja calculate de Excel).
|
||
• Dashboard — metrici 5 manageri + breakdown-uri + prop compliance (formule).
|
||
• Ferestre — AUTO-SCAN edge × durată × fiabilitate (python, valori statice).
|
||
Nicio fereastră hardcodată: grila se scanează la 15 min pe toate
|
||
cele 5 manageri și recomandările se DERIVĂ din datele curente.
|
||
• Toate ferestrele — grila completă scanată (toate ferestrele × 5 manageri) ca
|
||
tabel plat cu AutoFilter, ca să filtrezi/sortezi singur.
|
||
• date_grafic — sursa pentru curba de echitate din sheet-ul Ferestre.
|
||
|
||
Reruleaza prin refresh_dashboard.bat (sau direct):
|
||
python scripts/generate_dashboard.py
|
||
|
||
IMPORTANT: deschide și SALVEAZĂ backtest.xlsx în Excel cel puțin o dată după
|
||
ultima editare înainte de refresh. Scriptul citește valorile DEJA calculate de
|
||
Excel (R_/$_/Bal_/helpere). Dacă nu ai salvat în Excel, cache-ul de valori
|
||
lipsește și analiza iese goală.
|
||
|
||
Istoric: înainte erau două fișiere (Dashboard.xlsx + Ferestre_v2.xlsx) cu două
|
||
scannere de ferestre — unul pe formule (grila fixă din build_dashboard), altul pe
|
||
python cu ferestre A/B/W hardcodate de o analiză LLM anterioară. Ambele au fost
|
||
înlocuite cu un singur auto-scan generativ (grila la 15 min e suprasetul ambelor).
|
||
"""
|
||
|
||
import statistics
|
||
import random
|
||
from collections import defaultdict
|
||
from datetime import datetime, time, date, timedelta
|
||
from pathlib import Path
|
||
|
||
import openpyxl
|
||
from openpyxl import Workbook
|
||
from openpyxl.styles import Font, PatternFill, Alignment, Border, Side
|
||
from openpyxl.utils import get_column_letter
|
||
from openpyxl.chart import LineChart, Reference
|
||
from openpyxl.drawing.line import LineProperties
|
||
from openpyxl.chart.shapes import GraphicalProperties
|
||
|
||
from generate_template import (
|
||
build_config,
|
||
build_dashboard,
|
||
TRADES_HEADERS,
|
||
MAX_ROWS,
|
||
STRAT_KEYS,
|
||
STRAT_LABELS,
|
||
)
|
||
|
||
SRC = Path(__file__).resolve().parent.parent / "data" / "backtest.xlsx"
|
||
OUT = Path(__file__).resolve().parent.parent / "data" / "Dashboard.xlsx"
|
||
|
||
# Rândurile de input (galbene) din sheet-ul Config — singurele pe care le purtăm
|
||
# din workbook-ul real (Account, Lot, limite prop, calibrare $/punct). Restul
|
||
# celulelor Config sunt formule recreate de build_config().
|
||
CONFIG_INPUT_ROWS = [4, 5, 9, 10, 12, 14, 17, 19, 20, 21, 22, 25]
|
||
|
||
|
||
def read_config_inputs(ws_cfg) -> dict[int, object]:
|
||
"""Citește valorile din coloana B a sheet-ului Config (read-only)."""
|
||
vals: dict[int, object] = {}
|
||
for r, row in enumerate(
|
||
ws_cfg.iter_rows(min_row=1, max_row=40, min_col=2, max_col=2), start=1
|
||
):
|
||
# read_only poate întoarce EmptyCell (fără .value) pentru celule goale
|
||
vals[r] = getattr(row[0], "value", None)
|
||
return vals
|
||
|
||
|
||
def apply_config_inputs(wb: Workbook, cfg_inputs: dict[int, object]) -> None:
|
||
"""Suprascrie inputurile Config cu valorile reale ale lui Marius."""
|
||
ws = wb["Config"]
|
||
for r in CONFIG_INPUT_ROWS:
|
||
v = cfg_inputs.get(r)
|
||
if v is not None:
|
||
ws.cell(row=r, column=2, value=v)
|
||
|
||
|
||
def copy_trades_values(wb: Workbook, ws_src) -> None:
|
||
"""Creează un sheet Trades static (valori) în ordinea exactă TRADES_HEADERS.
|
||
|
||
Mapează după NUMELE coloanei din sursă, ca literele din COL să corespundă cu
|
||
ce așteaptă formulele/charturile din build_dashboard, indiferent de ordinea
|
||
fizică din backtest.xlsx.
|
||
"""
|
||
ws = wb.create_sheet("Trades", 1)
|
||
|
||
src_rows = ws_src.iter_rows(min_row=1, values_only=True)
|
||
src_hdr = next(src_rows)
|
||
src_idx = {name: i for i, name in enumerate(src_hdr) if name is not None}
|
||
|
||
# Header (necesar pentru titles_from_data al charturilor Bal_*/BalProp_*)
|
||
for col_idx, name in enumerate(TRADES_HEADERS, start=1):
|
||
ws.cell(row=1, column=col_idx, value=name)
|
||
|
||
# Date — rândurile 2..MAX_ROWS+1, ca rangurile Trades!$X$2:$X$501 să se alinieze
|
||
r_out = 2
|
||
for src_row in src_rows:
|
||
if r_out > MAX_ROWS + 1:
|
||
break
|
||
for col_idx, name in enumerate(TRADES_HEADERS, start=1):
|
||
si = src_idx.get(name)
|
||
if si is None or si >= len(src_row):
|
||
continue
|
||
val = src_row[si]
|
||
if val is not None:
|
||
ws.cell(row=r_out, column=col_idx, value=val)
|
||
r_out += 1
|
||
|
||
ws.sheet_state = "hidden" # snapshot intern; Dashboard e singurul vizibil util
|
||
|
||
|
||
# ===========================================================================
|
||
# FERESTRE — auto-scan edge × durată × fiabilitate (python, valori statice)
|
||
# ===========================================================================
|
||
#
|
||
# Înlocuiește vechiul Ferestre_v2.xlsx (ferestre A/B/W hardcodate). Aici NIMIC
|
||
# nu e fixat: grila de ferestre candidate se generează parametric la 15 min,
|
||
# fiecare e evaluată pe toate cele 5 manageri pe DATELE CURENTE, iar cele 3
|
||
# recomandări (edge/durată, robust, volum) se aleg după criterii obiective.
|
||
|
||
ROLUNI = {1: "Ian", 2: "Feb", 3: "Mar", 4: "Apr", 5: "Mai", 6: "Iun",
|
||
7: "Iul", 8: "Aug", 9: "Sep", 10: "Oct", 11: "Noi", 12: "Dec"}
|
||
|
||
# Parametrii grilei de scan (ora RO, în minute de la miezul nopții).
|
||
SCAN_START_MIN = 16 * 60 + 30 # 16:30 — cel mai devreme start testat
|
||
SCAN_LAST_START = 22 * 60 # 22:00 — cel mai târziu start testat
|
||
SCAN_HARD_END = 23 * 60 # 23:00 — capăt maxim al oricărei ferestre
|
||
SCAN_MIN_DUR = 45 # durată minimă fereastră (min)
|
||
SCAN_STEP = 15 # rezoluție (min) — suprasetul vechilor grile
|
||
SCAN_MIN_N = 30 # nr. minim de tranzacții ca o fereastră să conteze
|
||
NBOOT = 5000 # re-eșantioane bootstrap pentru CI
|
||
|
||
# Câte ferestre intră în validarea detaliată din sheet-ul Ferestre (BRUT + cele 3
|
||
# recomandări + restul = top-N profitabile auto-incluse). Pragurile decid ce e
|
||
# "profitabilă" pentru auto-includere — ridică-le ca să vezi mai puține/mai multe.
|
||
FERESTRE_MAX_VARIANTS = 20 # total maxim de rânduri validate (BRUT + recs + top scan)
|
||
FERESTRE_TOP_MIN_N = 40 # N minim pentru o fereastră suplimentară auto-inclusă
|
||
FERESTRE_TOP_MIN_EXPR = 0.10 # ExpR minim pentru "profitabilă" la auto-includere
|
||
|
||
|
||
def load_trades(path: Path) -> list[dict]:
|
||
"""Citește tranzacțiile cu outcome din backtest.xlsx (valori cache Excel)."""
|
||
wb = openpyxl.load_workbook(path, read_only=True, data_only=True)
|
||
ws = wb["Trades"]
|
||
it = ws.iter_rows(min_row=1, values_only=True)
|
||
hdr = next(it)
|
||
h = {x: i for i, x in enumerate(hdr) if x is not None}
|
||
rows: list[dict] = []
|
||
for row in it:
|
||
o = row[h["Outcome"]]
|
||
if o is None or str(o).strip() == "":
|
||
continue
|
||
t = row[h["Ora RO"]]
|
||
if isinstance(t, (int, float)):
|
||
hh = int(t); mm = round((t - hh) * 100); tt = time(hh, mm)
|
||
else:
|
||
p = str(t).split(":"); tt = time(int(p[0]), int(p[1]))
|
||
d = datetime.fromisoformat(str(row[h["Data"]])).date()
|
||
rec = {"d": d, "min": tt.hour * 60 + tt.minute,
|
||
"t": tt.strftime("%H:%M"), "dir": row[h["Direcție"]]}
|
||
for s in STRAT_KEYS:
|
||
rec["R_" + s] = row[h["R_" + s]]
|
||
rec["$_" + s] = row[h["$_" + s]]
|
||
rows.append(rec)
|
||
wb.close()
|
||
rows.sort(key=lambda r: (r["d"], r["min"]))
|
||
return rows
|
||
|
||
|
||
# ---------- engine ----------
|
||
def in_window(rows, s, e):
|
||
return [r for r in rows if s <= r["min"] < e]
|
||
|
||
|
||
def apply_filter(rows, f):
|
||
"""'toate' = tot; 'prima' = prima tranzacție cronologic din fiecare zi."""
|
||
if f == "toate":
|
||
return rows
|
||
byd = defaultdict(list)
|
||
for r in rows:
|
||
byd[r["d"]].append(r)
|
||
out = []
|
||
if f == "prima":
|
||
for d, rs in byd.items():
|
||
out.append(rs[0])
|
||
out.sort(key=lambda r: (r["d"], r["min"]))
|
||
return out
|
||
|
||
|
||
def metrics(rows, strat, acct, daily, maxl):
|
||
"""Metrici + flag breach prop pe un subset, pentru un manager."""
|
||
n = len(rows)
|
||
if n == 0:
|
||
return None
|
||
R = [r["R_" + strat] for r in rows]
|
||
D = [r["$_" + strat] for r in rows]
|
||
if any(x is None for x in R) or any(x is None for x in D):
|
||
return None
|
||
eq = acct; peak = acct; maxdd = 0.0
|
||
cur = None; daycum = 0.0; daymin = 0.0; dbreach = False; mbreach = False
|
||
for r in rows:
|
||
if r["d"] != cur:
|
||
cur = r["d"]; daycum = 0.0; daymin = 0.0
|
||
daycum += r["$_" + strat]; daymin = min(daymin, daycum)
|
||
if -daymin >= daily:
|
||
dbreach = True
|
||
eq += r["$_" + strat]; peak = max(peak, eq); maxdd = max(maxdd, peak - eq)
|
||
if peak - eq >= maxl:
|
||
mbreach = True
|
||
return dict(n=n, wr=sum(1 for x in R if x > 0) / n * 100,
|
||
exp=statistics.mean(R), totR=sum(R), totD=sum(D),
|
||
maxdd=maxdd, breach=dbreach or mbreach)
|
||
|
||
|
||
def monthly_expr(rows, strat) -> dict[str, float]:
|
||
bym = defaultdict(list)
|
||
for r in rows:
|
||
bym[f"{r['d']:%Y-%m}"].append(r["R_" + strat])
|
||
return {m: statistics.mean(v) for m, v in bym.items()}
|
||
|
||
|
||
def bootstrap(rows, strat, nboot=NBOOT, seed=12345):
|
||
"""Re-eșantionare cu înlocuire: CI 95% pentru ExpR."""
|
||
rnd = random.Random(seed)
|
||
R = [r["R_" + strat] for r in rows]
|
||
n = len(R)
|
||
if n == 0:
|
||
return dict(expR_lo=0.0, expR_hi=0.0, p_pos=0.0)
|
||
means = []
|
||
for _ in range(nboot):
|
||
means.append(sum(R[rnd.randrange(n)] for _ in range(n)) / n)
|
||
means.sort()
|
||
|
||
def pct(p):
|
||
return means[min(len(means) - 1, int(p * len(means)))]
|
||
|
||
return dict(expR_lo=pct(0.025), expR_hi=pct(0.975),
|
||
p_pos=sum(1 for x in means if x > 0) / nboot * 100)
|
||
|
||
|
||
# ---------- scan + recomandări ----------
|
||
def scan(T, acct, daily, maxl):
|
||
"""Scanează grila la 15 min × {toate, prima} × 5 manageri.
|
||
|
||
Pentru fiecare fereastră alege managerul cu cel mai bun ExpR dintre cei
|
||
care NU sparg contul prop pe acea fereastră. Întoarce lista de candidați.
|
||
"""
|
||
cands = []
|
||
for s in range(SCAN_START_MIN, SCAN_LAST_START + 1, SCAN_STEP):
|
||
for e in range(s + SCAN_MIN_DUR, SCAN_HARD_END + 1, SCAN_STEP):
|
||
for filt in ("toate", "prima"):
|
||
sel = apply_filter(in_window(T, s, e), filt)
|
||
if len(sel) < SCAN_MIN_N:
|
||
continue
|
||
best = None
|
||
for st in STRAT_KEYS:
|
||
m = metrics(sel, st, acct, daily, maxl)
|
||
if m is None or m["breach"]:
|
||
continue
|
||
if best is None or m["exp"] > best[1]["exp"]:
|
||
best = (st, m)
|
||
if best is None:
|
||
continue
|
||
st, m = best
|
||
mo = monthly_expr(sel, st)
|
||
cands.append(dict(
|
||
s=s, e=e, filt=filt, dur=e - s, strat=st,
|
||
n=m["n"], wr=m["wr"], exp=m["exp"], totD=m["totD"],
|
||
maxdd=m["maxdd"],
|
||
mpos=sum(1 for v in mo.values() if v > 0), mtot=len(mo),
|
||
))
|
||
return cands
|
||
|
||
|
||
def _pick(cands, predicate, key, taken):
|
||
"""Cel mai bun candidat (după key, desc) care trece predicate și nu e deja luat.
|
||
|
||
Dedup pe (start, end) — un geam apare o singură dată în recomandări, indiferent
|
||
de filtru/manager (ca ROBUST 2 să nu repete fereastra deja luată de EDGE etc.).
|
||
"""
|
||
pool = [c for c in cands
|
||
if predicate(c) and (c["s"], c["e"]) not in taken]
|
||
if not pool:
|
||
return None
|
||
best = max(pool, key=key)
|
||
taken.add((best["s"], best["e"]))
|
||
return best
|
||
|
||
|
||
# Constante de reglare a recomandărilor (modifică-le ca să schimbi criteriile).
|
||
EDGE_DURATIONS = [45, 60, 90] # paliere de durată (min): cea mai mică fereastră + 1h + 1h30
|
||
EDGE_MIN_EXPR = 0.10 # ExpR minim ca o fereastră "edge" să fie profitabilă
|
||
ROBUST_MIN_N = 40 # N minim pentru recomandările robust
|
||
ROBUST_GRADES = [ # grade de consistență (fracțiune minimă de luni pozitive)
|
||
("ROBUST 1 (toate lunile)", 1.00),
|
||
("ROBUST 2 (≥80% luni)", 0.80),
|
||
("ROBUST 3 (≥60% luni)", 0.60),
|
||
]
|
||
VOLUM_MIN_EXPR = 0.10 # ExpR minim ca o fereastră de volum să conteze
|
||
VOLUM_MIN_N = 60 # N minim ca "VOLUM compact" să fie relevant statistic
|
||
|
||
|
||
def recommend(cands):
|
||
"""Derivă recomandările din scan (etichete = ferestrele calculate, nimic fix).
|
||
|
||
Familii: EDGE (cea mai mică fereastră profitabilă, pe paliere de durată),
|
||
ROBUST 1/2/3 (grade de consistență lunară de la strict la lax) și VOLUM (cele
|
||
mai multe tranzacții profitabile + cea mai densă fereastră). Pragurile sunt în
|
||
constantele de mai sus. Dedup: fiecare fereastră apare o singură dată.
|
||
"""
|
||
taken: set = set()
|
||
out = []
|
||
|
||
def add(role, predicate, key):
|
||
c = _pick(cands, predicate, key, taken)
|
||
if c:
|
||
out.append((role, c))
|
||
|
||
# EDGE — cea mai mică perioadă profitabilă, pe paliere de durată (45min, 1h, 1h30).
|
||
for dur in EDGE_DURATIONS:
|
||
h, m = divmod(dur, 60)
|
||
lab = f"EDGE {h}h{m:02d}" if h else f"EDGE {m}min"
|
||
add(lab, lambda c, d=dur: c["dur"] == d and c["exp"] >= EDGE_MIN_EXPR,
|
||
key=lambda c: (c["exp"], c["n"]))
|
||
|
||
# ROBUST — grade de consistență lunară (strict → lax).
|
||
for role, frac in ROBUST_GRADES:
|
||
add(role,
|
||
lambda c, f=frac: c["n"] >= ROBUST_MIN_N and c["mpos"] / max(c["mtot"], 1) >= f,
|
||
key=lambda c: (c["exp"], c["n"]))
|
||
|
||
# VOLUM — cel mai mare volum (relevant statistic) + cea mai SCURTĂ fereastră
|
||
# care încă are volum relevant (N≥VOLUM_MIN_N).
|
||
add("VOLUM (max N)", lambda c: c["exp"] >= VOLUM_MIN_EXPR,
|
||
key=lambda c: (c["n"], -c["dur"]))
|
||
add(f"VOLUM compact (N≥{VOLUM_MIN_N})",
|
||
lambda c: c["exp"] >= VOLUM_MIN_EXPR and c["n"] >= VOLUM_MIN_N,
|
||
key=lambda c: (-c["dur"], c["n"]))
|
||
return out
|
||
|
||
|
||
# ---------- styles (sheet Ferestre) ----------
|
||
F_TITLE = Font(bold=True, size=14, color="1F4E78")
|
||
F_SUB = Font(bold=True, size=11, color="1F4E78")
|
||
F_HF = Font(bold=True, color="FFFFFF")
|
||
F_HFILL = PatternFill("solid", fgColor="1F4E78")
|
||
F_GOOD = PatternFill("solid", fgColor="C6EFCE")
|
||
F_WARN = PatternFill("solid", fgColor="FFEB9C")
|
||
F_BAD = PatternFill("solid", fgColor="FFC7CE")
|
||
F_GREY = PatternFill("solid", fgColor="F2F2F2")
|
||
F_CTR = Alignment(horizontal="center")
|
||
F_LEFT = Alignment(horizontal="left", wrap_text=True)
|
||
F_THIN = Border(left=Side(style="thin", color="BFBFBF"),
|
||
right=Side(style="thin", color="BFBFBF"),
|
||
top=Side(style="thin", color="BFBFBF"),
|
||
bottom=Side(style="thin", color="BFBFBF"))
|
||
|
||
|
||
def _fmt(m):
|
||
return f"{m // 60:02d}:{m % 60:02d}"
|
||
|
||
|
||
def _wlabel(cfg):
|
||
return "(fără fer.)" if cfg["e"] - cfg["s"] >= 1440 else f"{_fmt(cfg['s'])}-{_fmt(cfg['e'])}"
|
||
|
||
|
||
def _expcolor(e):
|
||
return F_GOOD if e >= 0.10 else (F_WARN if e >= 0 else F_BAD)
|
||
|
||
|
||
def build_ferestre(wb: Workbook, T: list[dict], acct, daily, maxl) -> None:
|
||
"""Construiește sheet-urile Ferestre + date_grafic din auto-scan."""
|
||
ws = wb.create_sheet("Ferestre")
|
||
ws.sheet_view.showGridLines = False
|
||
for i, w in enumerate([22, 14, 8, 9, 13, 7, 8, 9, 16, 9, 11, 10], 1):
|
||
ws.column_dimensions[get_column_letter(i)].width = w
|
||
|
||
d0 = min(r["d"] for r in T); d1 = max(r["d"] for r in T)
|
||
alld = sorted(set(r["d"] for r in T))
|
||
months = sorted({f"{r['d']:%Y-%m}" for r in T})
|
||
cut = alld[int(len(alld) * 0.70)]
|
||
tr = [r for r in T if r["d"] < cut]
|
||
te = [r for r in T if r["d"] >= cut]
|
||
|
||
cands = scan(T, acct, daily, maxl)
|
||
recs = recommend(cands)
|
||
|
||
# BRUT (fără fereastră) — referință, managerul cel mai bun pe tot setul.
|
||
brut_strat = max(
|
||
STRAT_KEYS,
|
||
key=lambda st: metrics(T, st, acct, daily, maxl)["exp"],
|
||
)
|
||
brut = dict(s=0, e=1440, filt="toate", strat=brut_strat)
|
||
|
||
# VARIANTE pentru tabel + validări: BRUT + 3 recomandări + top scan suplimentar.
|
||
variants = [("BRUT (ref.)", brut, F_GREY)]
|
||
taken = {(0, 1440)}
|
||
for role, c in recs:
|
||
variants.append((role, c, F_GOOD))
|
||
taken.add((c["s"], c["e"]))
|
||
for c in sorted(cands, key=lambda c: c["exp"], reverse=True):
|
||
if len(variants) >= FERESTRE_MAX_VARIANTS:
|
||
break
|
||
if (c["s"], c["e"]) in taken:
|
||
continue
|
||
if c["n"] < FERESTRE_TOP_MIN_N or c["exp"] < FERESTRE_TOP_MIN_EXPR:
|
||
continue
|
||
taken.add((c["s"], c["e"]))
|
||
variants.append(("top scan", c, None))
|
||
|
||
state = {"R": 1}
|
||
|
||
def put(r, c, v, font=None, fill=None, fmtn=None, align=None, border=False):
|
||
cell = ws.cell(row=r, column=c, value=v)
|
||
if font: cell.font = font
|
||
if fill: cell.fill = fill
|
||
if fmtn: cell.number_format = fmtn
|
||
if align: cell.alignment = align
|
||
if border: cell.border = F_THIN
|
||
return cell
|
||
|
||
def title(txt):
|
||
put(state["R"], 1, txt, F_SUB); state["R"] += 1
|
||
|
||
def headers(hs):
|
||
for j, hh in enumerate(hs, 1):
|
||
put(state["R"], j, hh, F_HF, F_HFILL, align=F_CTR, border=True)
|
||
state["R"] += 1
|
||
|
||
def row(vals, fills=None, fmts=None):
|
||
for j, v in enumerate(vals, 1):
|
||
f = fills[j - 1] if fills else None
|
||
nf = fmts[j - 1] if fmts else None
|
||
put(state["R"], j, v, fill=f, fmtn=nf, border=True,
|
||
align=F_CTR if j > 1 else F_LEFT)
|
||
state["R"] += 1
|
||
|
||
def note(txt):
|
||
put(state["R"], 1, txt, align=F_LEFT); state["R"] += 1
|
||
|
||
def blank():
|
||
state["R"] += 1
|
||
|
||
def pad(vals, fills, fmts, n=12):
|
||
while len(vals) < n:
|
||
vals.append(""); fills.append(None); fmts.append(None)
|
||
return vals, fills, fmts
|
||
|
||
def msel(rows, cfg):
|
||
return metrics(apply_filter(in_window(rows, cfg["s"], cfg["e"]), cfg["filt"]),
|
||
cfg["strat"], acct, daily, maxl)
|
||
|
||
# ---- titlu (fără note de intro — tabelul de mai jos vorbește) ----
|
||
put(state["R"], 1, "FERESTRE — auto-scan edge × durată × fiabilitate", F_TITLE)
|
||
state["R"] += 1
|
||
blank()
|
||
|
||
# ---- TABEL UNIC ----
|
||
title("TABEL — ferestrele de top din scan (rol + manager cel mai bun)")
|
||
note(f"Roluri auto-derivate (nimic hardcodat): BRUT = fără fereastră (referință) · "
|
||
f"EDGE 45min/1h/1h30 = cea mai profitabilă fereastră la EXACT acea durată (45min = cea mai scurtă perioadă profitabilă) · "
|
||
f"ROBUST 1/2/3 = cel mai bun ExpR pozitiv în toate / ≥80% / ≥60% din luni (N≥{ROBUST_MIN_N}) · "
|
||
f"VOLUM (max N) = cel mai mare volum profitabil (relevant statistic) · VOLUM compact = cea mai scurtă fereastră cu N≥{VOLUM_MIN_N} · "
|
||
f"top scan = restul ferestrelor profitabile (N≥{FERESTRE_TOP_MIN_N}, ExpR≥{FERESTRE_TOP_MIN_EXPR:.2f}, non-breach) după ExpR, până la {FERESTRE_MAX_VARIANTS} total. "
|
||
f"CI 95% = interval bootstrap · OOS = ExpR pe ultimele ~30% zile · Manager = cel mai bun dintre cei 5. "
|
||
"Vezi 'Toate ferestrele' pentru grila completă filtrabilă.")
|
||
headers(["Rol", "Fereastră", "Durată", "Filtru", "Manager", "N", "WR%",
|
||
"ExpR", "CI 95% ExpR", "OOS", "$ total", "maxDD$"])
|
||
for role, cfg, fill in variants:
|
||
sel = apply_filter(in_window(T, cfg["s"], cfg["e"]), cfg["filt"])
|
||
m = metrics(sel, cfg["strat"], acct, daily, maxl)
|
||
b = bootstrap(sel, cfg["strat"])
|
||
oosm = msel(te, cfg); oosx = oosm["exp"] if oosm else 0.0
|
||
dd = cfg["e"] - cfg["s"]
|
||
durs = "—" if dd >= 1440 else f"{dd // 60}h{dd % 60:02d}"
|
||
row([role, _wlabel(cfg), durs, cfg["filt"], STRAT_LABELS[cfg["strat"]],
|
||
m["n"], m["wr"], round(m["exp"], 3),
|
||
f"[{b['expR_lo']:+.2f};{b['expR_hi']:+.2f}]", round(oosx, 3),
|
||
round(m["totD"]), round(m["maxdd"])],
|
||
fills=[fill, None, None, None, None, None, None, None, None,
|
||
_expcolor(oosx), None, None],
|
||
fmts=[None, None, None, None, None, "0", '0.0"%"', "0.000", None,
|
||
"0.000", "$#,##0", "$#,##0"])
|
||
blank()
|
||
|
||
# ---- explicații validări (definiții, NU concluzii) ----
|
||
title("CE ÎNSEAMNĂ VALIDĂRILE")
|
||
note("• Forward 1 (LUNAR): ExpR în fiecare lună separat. Pozitiv în toate = edge constant, nu noroc concentrat. N mic/lună → o tranzacție mișcă mult media.")
|
||
note("• Forward 2 (TRAIN/TEST): primele 70% din zile = train, ultimele 30% = test (nevăzut la alegere). ExpR test ≈ train → robust; mult mai mic/negativ → overfit.")
|
||
note("• Walk-forward (3 FELII): perioada în 3 bucăți cronologice. O regulă bună rămâne pozitivă în toate trei, nu doar la început.")
|
||
note("• Culori: VERDE ≥0.10R · GALBEN 0–0.10R · ROȘU negativ · gol = nicio tranzacție în acea felie/lună.")
|
||
blank()
|
||
|
||
# ---- FORWARD 1 — LUNAR ----
|
||
mlabels = [ROLUNI[int(m[5:7])] for m in months]
|
||
title("FORWARD 1 — consistență LUNARĂ (ExpR pe fiecare lună)")
|
||
headers(["Variantă", "Fereastră"] + mlabels)
|
||
for role, cfg, fill in variants:
|
||
sel = apply_filter(in_window(T, cfg["s"], cfg["e"]), cfg["filt"])
|
||
bym = defaultdict(list)
|
||
for r in sel:
|
||
bym[f"{r['d']:%Y-%m}"].append(r["R_" + cfg["strat"]])
|
||
vals = [role, _wlabel(cfg)]; fills = [fill, None]; fmts = [None, None]
|
||
for m in months:
|
||
rr = bym.get(m, [])
|
||
if rr:
|
||
e = statistics.mean(rr)
|
||
vals.append(round(e, 3)); fills.append(_expcolor(e)); fmts.append("0.000")
|
||
else:
|
||
vals.append(""); fills.append(F_GREY); fmts.append(None)
|
||
row(*pad(vals, fills, fmts, n=2 + len(months)))
|
||
blank()
|
||
|
||
# ---- FORWARD 2 — TRAIN/TEST ----
|
||
title("FORWARD 2 — TRAIN/TEST 70/30")
|
||
note(f"Train: {tr[0]['d']:%d.%m}–{cut:%d.%m} · Test/OOS: {cut:%d.%m}–{d1:%d.%m}. "
|
||
"Verde la ExpR test = edge-ul a ținut pe date nevăzute. Δ≈0 sau pozitiv = stabil; foarte negativ = overfit.")
|
||
headers(["Variantă", "Fereastră", "N train", "ExpR train", "N test",
|
||
"ExpR test (OOS)", "Δ (test−train)"])
|
||
for role, cfg, fill in variants:
|
||
mtr = msel(tr, cfg); mte = msel(te, cfg)
|
||
etr = mtr["exp"] if mtr else 0.0; ete = mte["exp"] if mte else 0.0
|
||
ntr = mtr["n"] if mtr else 0; nte = mte["n"] if mte else 0
|
||
row([role, _wlabel(cfg), ntr, round(etr, 3), nte, round(ete, 3),
|
||
round(ete - etr, 3)],
|
||
fills=[fill, None, None, None, None, _expcolor(ete), None],
|
||
fmts=[None, None, "0", "0.000", "0", "0.000", "0.000"])
|
||
blank()
|
||
|
||
# ---- WALK-FORWARD — 3 FELII ----
|
||
n3 = len(alld) // 3
|
||
P = [set(alld[:n3]), set(alld[n3:2 * n3]), set(alld[2 * n3:])]
|
||
pr = [(alld[0], alld[n3 - 1]), (alld[n3], alld[2 * n3 - 1]), (alld[2 * n3], alld[-1])]
|
||
title("WALK-FORWARD — edge pe 3 FELII cronologice")
|
||
note("P1=%s–%s · P2=%s–%s · P3=%s–%s. Pozitiv (verde) în toate trei = edge stabil în timp." % (
|
||
pr[0][0].strftime("%d.%m"), pr[0][1].strftime("%d.%m"),
|
||
pr[1][0].strftime("%d.%m"), pr[1][1].strftime("%d.%m"),
|
||
pr[2][0].strftime("%d.%m"), pr[2][1].strftime("%d.%m")))
|
||
headers(["Variantă", "Fereastră", "P1 ExpR", "P2 ExpR", "P3 ExpR", "N total"])
|
||
for role, cfg, fill in variants:
|
||
sel = apply_filter(in_window(T, cfg["s"], cfg["e"]), cfg["filt"])
|
||
vals = [role, _wlabel(cfg)]; fills = [fill, None]; fmts = [None, None]
|
||
for ps in P:
|
||
rr = [r for r in sel if r["d"] in ps]
|
||
if rr:
|
||
e = statistics.mean([x["R_" + cfg["strat"]] for x in rr])
|
||
vals.append(round(e, 3)); fills.append(_expcolor(e)); fmts.append("0.000")
|
||
else:
|
||
vals.append(""); fills.append(F_GREY); fmts.append(None)
|
||
vals.append(len(sel)); fills.append(None); fmts.append("0")
|
||
row(vals, fills, fmts)
|
||
blank()
|
||
|
||
# ---- CALENDAR (FOMC/NFP) ----
|
||
title("CALENDAR EVENIMENTE — influență?")
|
||
yrs = sorted({d.year for d in alld})
|
||
|
||
def first_fri(y, mo):
|
||
d = date(y, mo, 1)
|
||
while d.weekday() != 4:
|
||
d += timedelta(days=1)
|
||
return d
|
||
|
||
NFP = {first_fri(y, mo) for y in yrs for mo in range(1, 13)
|
||
if date(y, mo, 1) <= d1}
|
||
headers(["Grup", "N", "WR%", "ExpR (best mgr)"])
|
||
|
||
def grp(rows):
|
||
if not rows:
|
||
return 0, 0, 0
|
||
best = max(STRAT_KEYS, key=lambda st: statistics.mean([r["R_" + st] for r in rows]))
|
||
Rm = [r["R_" + best] for r in rows]
|
||
return len(Rm), sum(1 for x in Rm if x > 0) / len(Rm) * 100, statistics.mean(Rm)
|
||
|
||
for label, rows in (("Zile NFP (prima vineri)", [r for r in T if r["d"] in NFP]),
|
||
("Restul zilelor", [r for r in T if r["d"] not in NFP])):
|
||
n, wr, ex = grp(rows)
|
||
row([label, n, wr, round(ex, 3)],
|
||
fmts=[None, "0", '0.0"%"', "0.000"])
|
||
note("Prea puține zile de eveniment pentru o regulă de news-filter; aici doar ca verificare că nu strică edge-ul.")
|
||
blank()
|
||
|
||
# ---- GRAFIC — curbă de echitate pe primele 2 recomandări ----
|
||
def _find(role_name):
|
||
for role, cfg, _ in variants:
|
||
if role == role_name:
|
||
return (role, cfg)
|
||
return None
|
||
|
||
chart_variants = [v for v in (_find(ROBUST_GRADES[0][0]), _find("VOLUM (max N)")) if v]
|
||
if len(chart_variants) < 2: # fallback: primele 2 recomandări non-BRUT
|
||
for role, cfg, _ in variants:
|
||
if role != "BRUT (ref.)" and (role, cfg) not in chart_variants:
|
||
chart_variants.append((role, cfg))
|
||
if len(chart_variants) == 2:
|
||
break
|
||
chart_variants = chart_variants[:2]
|
||
if len(chart_variants) == 2:
|
||
(r1, c1), (r2, c2) = chart_variants
|
||
l1 = f"{r1} {_fmt(c1['s'])}-{_fmt(c1['e'])}"
|
||
l2 = f"{r2} {_fmt(c2['s'])}-{_fmt(c2['e'])}"
|
||
title(f"GRAFIC — curbă de echitate ($ cumulativ): {l1} vs {l2}")
|
||
note("Aliniate pe dată. Compară câștigul și 'netezimea' celor mai bune două recomandări.")
|
||
chart_anchor = f"A{state['R'] + 1}"
|
||
|
||
def daily_sum(cfg):
|
||
sel = apply_filter(in_window(T, cfg["s"], cfg["e"]), cfg["filt"])
|
||
byd = defaultdict(float)
|
||
for r in sel:
|
||
byd[r["d"]] += r["$_" + cfg["strat"]]
|
||
return byd
|
||
|
||
cumA = daily_sum(c1); cumB = daily_sum(c2)
|
||
ds = wb.create_sheet("date_grafic")
|
||
ds.append(["Data", l1, l2])
|
||
accA = 0.0; accB = 0.0
|
||
for d in alld:
|
||
accA += cumA.get(d, 0.0); accB += cumB.get(d, 0.0)
|
||
ds.append([d, round(accA), round(accB)])
|
||
nrows = len(alld)
|
||
for rr in range(2, nrows + 2):
|
||
ds.cell(row=rr, column=1).number_format = "dd.mm"
|
||
|
||
chart = LineChart()
|
||
chart.title = f"Curbă de echitate ($ cumulativ) — {l1} vs {l2}"
|
||
chart.style = 2
|
||
chart.height = 9.5; chart.width = 24
|
||
chart.y_axis.title = "$ cumulativ (cont prop)"
|
||
chart.x_axis.title = "Data"
|
||
chart.x_axis.number_format = "dd.mm"
|
||
chart.x_axis.majorTimeUnit = "days"
|
||
chart.x_axis.delete = False
|
||
chart.y_axis.delete = False
|
||
data = Reference(ds, min_col=2, max_col=3, min_row=1, max_row=nrows + 1)
|
||
cats = Reference(ds, min_col=1, min_row=2, max_row=nrows + 1)
|
||
chart.add_data(data, titles_from_data=True)
|
||
chart.set_categories(cats)
|
||
for s, color in zip(chart.series, ("2E7D32", "1F4E78")):
|
||
s.graphicalProperties = GraphicalProperties()
|
||
s.graphicalProperties.line = LineProperties(solidFill=color, w=20000)
|
||
s.smooth = False
|
||
ws.add_chart(chart, chart_anchor)
|
||
ds.column_dimensions["A"].width = 11
|
||
for col in ("B", "C"):
|
||
ds.column_dimensions[col].width = 18
|
||
|
||
|
||
def build_scan_table(wb: Workbook, T: list[dict], acct, daily, maxl) -> int:
|
||
"""Sheet cu TOATE ferestrele scanate × toți 5 managerii — tabel plat filtrabil.
|
||
|
||
Spre deosebire de sheet-ul Ferestre (care arată doar top-ul + validările), aici
|
||
e grila completă cu AutoFilter, ca Marius să filtreze singur după N, ExpR,
|
||
breach, manager, durată etc. Sortat descrescător după ExpR.
|
||
"""
|
||
ws = wb.create_sheet("Toate ferestrele")
|
||
ws.sheet_view.showGridLines = False
|
||
|
||
alld = sorted(set(r["d"] for r in T))
|
||
cut = alld[int(len(alld) * 0.70)]
|
||
te = [r for r in T if r["d"] >= cut]
|
||
|
||
head = ["Start", "End", "Durată min", "Filtru", "Manager", "N", "WR%",
|
||
"ExpR", "OOS ExpR", "$ total", "maxDD$", "Breach", "Luni+", "Luni tot"]
|
||
for j, h in enumerate(head, 1):
|
||
c = ws.cell(1, j, h)
|
||
c.font = F_HF; c.fill = F_HFILL; c.alignment = F_CTR; c.border = F_THIN
|
||
|
||
rows = []
|
||
for s in range(SCAN_START_MIN, SCAN_LAST_START + 1, SCAN_STEP):
|
||
for e in range(s + SCAN_MIN_DUR, SCAN_HARD_END + 1, SCAN_STEP):
|
||
for filt in ("toate", "prima"):
|
||
sel = apply_filter(in_window(T, s, e), filt)
|
||
if not sel:
|
||
continue
|
||
seloos = apply_filter(in_window(te, s, e), filt)
|
||
for st in STRAT_KEYS:
|
||
m = metrics(sel, st, acct, daily, maxl)
|
||
if m is None:
|
||
continue
|
||
mo = monthly_expr(sel, st)
|
||
moos = metrics(seloos, st, acct, daily, maxl)
|
||
rows.append((s, e, filt, st, m, mo, moos))
|
||
|
||
rows.sort(key=lambda x: x[4]["exp"], reverse=True)
|
||
|
||
r = 2
|
||
for s, e, filt, st, m, mo, moos in rows:
|
||
oos = round(moos["exp"], 3) if moos else ""
|
||
vals = [_fmt(s), _fmt(e), e - s, filt, STRAT_LABELS[st], m["n"],
|
||
m["wr"], round(m["exp"], 3), oos, round(m["totD"]),
|
||
round(m["maxdd"]), "DA" if m["breach"] else "NU",
|
||
sum(1 for v in mo.values() if v > 0), len(mo)]
|
||
for j, v in enumerate(vals, 1):
|
||
cell = ws.cell(r, j, v)
|
||
cell.border = F_THIN
|
||
cell.alignment = F_LEFT if j in (4, 5) else F_CTR
|
||
ws.cell(r, 6).number_format = "0"
|
||
ws.cell(r, 7).number_format = '0.0"%"'
|
||
ws.cell(r, 8).number_format = "+0.000;-0.000;0.000"
|
||
if moos:
|
||
ws.cell(r, 9).number_format = "+0.000;-0.000;0.000"
|
||
ws.cell(r, 10).number_format = "$#,##0"
|
||
ws.cell(r, 11).number_format = "$#,##0"
|
||
ws.cell(r, 8).fill = _expcolor(m["exp"])
|
||
if m["breach"]:
|
||
ws.cell(r, 12).fill = F_BAD
|
||
r += 1
|
||
|
||
last = r - 1
|
||
if last >= 2:
|
||
ws.auto_filter.ref = f"A1:N{last}"
|
||
ws.freeze_panes = "A2"
|
||
for i, w in enumerate([7, 7, 11, 8, 14, 6, 8, 9, 10, 11, 11, 8, 7, 8], 1):
|
||
ws.column_dimensions[get_column_letter(i)].width = w
|
||
return len(rows)
|
||
|
||
|
||
def main() -> int:
|
||
if not SRC.exists():
|
||
print(f"EROARE: nu găsesc {SRC}")
|
||
return 1
|
||
|
||
wb_src = openpyxl.load_workbook(SRC, read_only=True, data_only=True)
|
||
if "Trades" not in wb_src.sheetnames or "Config" not in wb_src.sheetnames:
|
||
print("EROARE: backtest.xlsx nu are sheet-urile Trades + Config.")
|
||
return 1
|
||
|
||
cfg_inputs = read_config_inputs(wb_src["Config"])
|
||
|
||
wb = Workbook()
|
||
wb.remove(wb.active)
|
||
build_config(wb) # Config la index 0 (cu formule)
|
||
apply_config_inputs(wb, cfg_inputs)
|
||
copy_trades_values(wb, wb_src["Trades"]) # Trades static la index 1 (ascuns)
|
||
build_dashboard(wb) # Dashboard la index 2 — formule
|
||
wb_src.close()
|
||
|
||
# Ferestre — auto-scan (limitele prop din Config, cu fallback la default).
|
||
acct = cfg_inputs.get(9) or 50000.0
|
||
daily = acct * (cfg_inputs.get(12) or 4.0) / 100.0
|
||
maxl = acct * (cfg_inputs.get(14) or 7.0) / 100.0
|
||
T = load_trades(SRC)
|
||
nscan = 0
|
||
if T:
|
||
build_ferestre(wb, T, acct, daily, maxl)
|
||
nscan = build_scan_table(wb, T, acct, daily, maxl)
|
||
else:
|
||
print("ATENȚIE: niciun trade cu outcome în cache — sar peste sheet-urile de analiză.")
|
||
|
||
# Ordine logică: rezumat (Ferestre) → grila completă → datele chart-ului.
|
||
order = ["Config", "Trades", "Dashboard", "Ferestre",
|
||
"Toate ferestrele", "date_grafic"]
|
||
wb._sheets.sort(key=lambda s: order.index(s.title) if s.title in order else 99)
|
||
|
||
wb.active = wb.sheetnames.index("Dashboard")
|
||
wb.save(OUT)
|
||
print(f"Scris {OUT} ({len(T)} tranzacții, {nscan} rânduri scan, "
|
||
f"sheet-uri: {wb.sheetnames})")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|