# -*- 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())