"""W7-TEST shared engine: reproducible vectorbtpro runs on Dukascopy H1 Bid CSVs. Data: assets/strategy-tests/_data/{eurusd,eurjpy}.csv (timestamp ms, OHLC bid, UTC+2 wall clock). Raw data is read-only and never copied or committed. Execution requirement: the caller (test.py) MUST import vectorbtpro and run under Windows ``py -3.11`` where vectorbtpro is installed. This module imports vectorbtpro inside run_portfolio() and raises if it is missing or if the portfolio trade count does not reconcile with the deterministic ledger. Conventions (explicit, no future leakage): - Signals use closed-bar data only (bar i uses inputs at index <= i). - Entry is at the NEXT bar's open (index i+1). No intrabar signal entry. - One position per symbol; signals while a position is open are ignored. - TP/SL are fixed at order time: risk = |entry - SL|, TP = entry +/- 1.5*risk. - Each bar from the entry bar on, bid high/low is checked against TP/SL. A bar touching both levels resolves STOP-FIRST (conservative); the trade is flagged (both_hit=1) and such trades are counted. - Costs are fixed assumptions (bid-only data cannot evidence them): EURUSD spread 1.0 pip, EURJPY spread 1.2 pip, $7/lot round trip at 0.01 lot = $0.07/trade. Half-spread adverse on every fill; commission converted to price units per side. EURJPY commission needs USD->JPY conversion; a fixed USDJPY=140 assumption is used and recorded (simplification). - IS/OOS split is predeclared: 2024-01-01 00:00 UTC+2, assigned by exit time. """ import os TARGET_R = 1.5 SPLIT_ISO = "2024-01-01T00:00:00+02:00" SPLIT_MS = 1704060000000 # 2024-01-01 00:00 UTC+2 WARMUP_BARS = 300 # indicator warmup; signals only at index >= WARMUP_BARS LOT = 0.01 COMM_USD_RT_PER_LOT = 7.0 USDJPY_FIXED = 140.0 # simplification: fixed conversion for EURJPY commission SYM = { "EURUSD": {"pip": 0.0001, "spread_pips": 1.0}, "EURJPY": {"pip": 0.01, "spread_pips": 1.2}, } DATA_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "_data") def load(symbol): """Read working-copy CSV. Returns DataFrame with ms timestamp + OHLC.""" import pandas as pd path = os.path.join(DATA_DIR, symbol.lower() + ".csv") df = pd.read_csv(path) assert list(df.columns) == ["timestamp", "open", "high", "low", "close"], df.columns assert df["timestamp"].is_monotonic_increasing, "timestamps not monotonic" assert not df["timestamp"].duplicated().any(), "duplicate bars" assert (df["high"] >= df[["open", "low", "close"]].max(axis=1)).all() assert (df["low"] <= df[["open", "high", "close"]].min(axis=1)).all() return df def ema(s, n): return s.ewm(span=n, min_periods=n, adjust=False).mean() def atr_wilder(df, n): import pandas as pd pc = df["close"].shift(1) tr = pd.concat( [df["high"] - df["low"], (df["high"] - pc).abs(), (df["low"] - pc).abs()], axis=1, ).max(axis=1) return tr.ewm(alpha=1.0 / n, min_periods=n, adjust=False).mean() def rsi_wilder(close, n): d = close.diff() gain = d.clip(lower=0.0) loss = -d.clip(upper=0.0) ag = gain.ewm(alpha=1.0 / n, min_periods=n, adjust=False).mean() al = loss.ewm(alpha=1.0 / n, min_periods=n, adjust=False).mean() return 100.0 - 100.0 / (1.0 + ag / al) def adx_wilder(df, n): import numpy as np up = df["high"].diff() dn = -df["low"].diff() plus_dm = up.where((up > dn) & (up > 0), 0.0) minus_dm = dn.where((dn > up) & (dn > 0), 0.0) pc = df["close"].shift(1) tr = np.maximum( df["high"] - df["low"], np.maximum((df["high"] - pc).abs(), (df["low"] - pc).abs()), ) tr_s = tr.ewm(alpha=1.0 / n, min_periods=n, adjust=False).mean() plus_di = 100.0 * plus_dm.ewm(alpha=1.0 / n, min_periods=n, adjust=False).mean() / tr_s minus_di = 100.0 * minus_dm.ewm(alpha=1.0 / n, min_periods=n, adjust=False).mean() / tr_s dx = 100.0 * (plus_di - minus_di).abs() / (plus_di + minus_di) return dx.ewm(alpha=1.0 / n, min_periods=n, adjust=False).mean() def supertrend(df, period, mult): """Classic causal Supertrend on median price. Returns (line, direction).""" import numpy as np h = df["high"].to_numpy() low = df["low"].to_numpy() close = df["close"].to_numpy() basis = (h + low) / 2.0 atr = atr_wilder(df, period).to_numpy() basic_ub = basis + mult * atr basic_lb = basis - mult * atr n = len(df) fub = np.full(n, np.nan) flb = np.full(n, np.nan) st = np.full(n, np.nan) first = int(np.where(~np.isnan(basic_ub))[0][0]) fub[first] = basic_ub[first] flb[first] = basic_lb[first] st[first] = fub[first] for i in range(first + 1, n): if np.isnan(basic_ub[i]): fub[i] = fub[i - 1] flb[i] = flb[i - 1] st[i] = st[i - 1] continue fub[i] = basic_ub[i] if (basic_ub[i] < fub[i - 1] or close[i - 1] > fub[i - 1]) else fub[i - 1] flb[i] = basic_lb[i] if (basic_lb[i] > flb[i - 1] or close[i - 1] < flb[i - 1]) else flb[i - 1] if st[i - 1] == fub[i - 1]: st[i] = fub[i] if close[i] <= fub[i] else flb[i] else: st[i] = flb[i] if close[i] >= flb[i] else fub[i] direction = np.where(st == flb, 1, -1) import pandas as pd return pd.Series(st, index=df.index), pd.Series(direction, index=df.index) def compute_all(df, p): """All indicators for a variant param set. Every output is causal.""" ind = {} ind["ema200"] = ema(df["close"], 200) ind["ema21"] = ema(df["close"], 21) ind["atr14"] = atr_wilder(df, 14) ind["rsi"] = rsi_wilder(df["close"], p["rsi_period"]) ind["adx"] = adx_wilder(df, p["adx_period"]) st_line, st_dir = supertrend(df, p["st_period"], p["st_mult"]) ind["st_line"] = st_line ind["st_dir"] = st_dir return ind def _gate(ind, i): if i < WARMUP_BARS: return False return all(v.iloc[i] == v.iloc[i] for v in ind.values()) # all finite def build_signals(strategy, df, ind, p): """Signal bars (closed-bar index i) with direction and raw SL level.""" import numpy as np n = len(df) close = df["close"].to_numpy() rsi = ind["rsi"].to_numpy() st_dir = ind["st_dir"].to_numpy() st_line = ind["st_line"].to_numpy() ema200 = ind["ema200"].to_numpy() ema21 = ind["ema21"].to_numpy() adx = ind["adx"].to_numpy() atr14 = ind["atr14"].to_numpy() sigs = [] for i in range(max(WARMUP_BARS, 1), n - 1): # need bar i+1 to enter if not _gate(ind, i): continue if strategy == "tr-supertrend": prev, cur = st_dir[i - 1], st_dir[i] if cur == 1 and prev == -1 and close[i] > ema200[i]: d = 1 elif cur == -1 and prev == 1 and close[i] < ema200[i]: d = -1 else: continue sl = st_line[i] - p["stop_buf_atr"] * atr14[i] if d == 1 else st_line[i] + p["stop_buf_atr"] * atr14[i] elif strategy == "tr-rsi-supertrend": if p["rsi_mode"] == "cross": up = (rsi[i - 1] <= 70.0) and (rsi[i] > 70.0) dn = (rsi[i - 1] >= 30.0) and (rsi[i] < 30.0) else: # hold: level only, no cross required up = rsi[i] > 70.0 dn = rsi[i] < 30.0 if up and st_dir[i] == 1: d = 1 elif dn and st_dir[i] == -1: d = -1 else: continue sl = st_line[i] - p["stop_buf_atr"] * atr14[i] if d == 1 else st_line[i] + p["stop_buf_atr"] * atr14[i] elif strategy == "tr-rsi-adx": if p["rsi_mode"] == "cross": up = (rsi[i - 1] <= 70.0) and (rsi[i] > 70.0) dn = (rsi[i - 1] >= 30.0) and (rsi[i] < 30.0) else: up = rsi[i] > 70.0 dn = rsi[i] < 30.0 adx_ok = adx[i] > p["adx_thr"] if p.get("adx_rising"): adx_ok = adx_ok and adx[i] > adx[i - 1] if up and close[i] > ema200[i] and adx_ok: d = 1 elif dn and close[i] < ema200[i] and adx_ok: d = -1 else: continue sl = ema21[i] - p["stop_buf_atr"] * atr14[i] if d == 1 else ema21[i] + p["stop_buf_atr"] * atr14[i] else: raise ValueError(strategy) sigs.append((i, d, float(sl))) return sigs def cost_model(symbol): """Fixed assumed costs in price units. Returns dict.""" spec = SYM[symbol] pip = spec["pip"] hs = spec["spread_pips"] * pip / 2.0 # half-spread adverse per fill comm_rt_usd = COMM_USD_RT_PER_LOT * LOT # $0.07 per trade if symbol == "EURUSD": # $0.035 per side / ($0.10 per pip at 0.01 lot) = 0.35 pip comm_side_px = 0.35 * pip else: # EURJPY: USD commission -> JPY at fixed USDJPY assumption jpy_per_pip_per_lot = 1000.0 # 1 pip (0.01 JPY) on 1000 units comm_side_pips = (comm_rt_usd / 2.0) * USDJPY_FIXED / jpy_per_pip_per_lot comm_side_px = comm_side_pips * pip return {"half_spread": hs, "comm_side": comm_side_px, "spread_pips": spec["spread_pips"]} def walk(df, sigs, p, cost, atr14): """Deterministic trade walk. Returns (ledger rows, n_skipped_degen, n_skipped_wide).""" n = len(df) o = df["open"].to_numpy() h = df["high"].to_numpy() low = df["low"].to_numpy() c = df["close"].to_numpy() ts = df["timestamp"].to_numpy() rows = [] flat_from = 0 skipped_degen = 0 skipped_wide = 0 for sig_i, d, sl in sigs: if sig_i < flat_from: continue # one position: ignore while open e = sig_i + 1 entry_raw = float(o[e]) risk = (entry_raw - sl) if d == 1 else (sl - entry_raw) if not risk > 0: skipped_degen += 1 continue if p.get("wide_skip_atr") and risk > p["wide_skip_atr"] * float(atr14[sig_i]): skipped_wide += 1 continue tp = entry_raw + TARGET_R * risk if d == 1 else entry_raw - TARGET_R * risk hs = cost["half_spread"] entry_fill = entry_raw + hs if d == 1 else entry_raw - hs reason, exit_j, exit_level, both = "END", n - 1, float(c[n - 1]), 0 for j in range(e, n): if d == 1: sl_hit = low[j] <= sl tp_hit = h[j] >= tp else: sl_hit = h[j] >= sl tp_hit = low[j] <= tp if sl_hit and tp_hit: both = 1 # ambiguous bar: conservative stop-first if sl_hit: reason, exit_j, exit_level = "SL", j, sl break if tp_hit: reason, exit_j, exit_level = "TP", j, tp break exit_fill = exit_level - hs if d == 1 else exit_level + hs gross = (exit_level - entry_raw) * d net = (exit_fill - entry_fill) * d - 2.0 * cost["comm_side"] rows.append( { "sig_bar": sig_i, "entry_bar": e, "exit_bar": exit_j, "direction": "long" if d == 1 else "short", "signal_time": int(ts[sig_i]), "entry_time": int(ts[e]), "exit_time": int(ts[exit_j]), "entry_raw": entry_raw, "sl": sl, "tp": tp, "risk": risk, "exit_level": exit_level, "entry_fill": entry_fill, "exit_fill": exit_fill, "exit_reason": reason, "both_hit": both, "cost_px": gross - net, "net_px": net, "net_R": net / risk, } ) flat_from = exit_j # next signal allowed once this position closed return rows, skipped_degen, skipped_wide def summarize(rows, symbol, cost, period_start_ms, period_end_ms): import numpy as np deci = [r for r in rows if r["exit_reason"] in ("TP", "SL")] n_tp = sum(1 for r in rows if r["exit_reason"] == "TP") n_sl = sum(1 for r in rows if r["exit_reason"] == "SL") n_end = sum(1 for r in rows if r["exit_reason"] == "END") win_rate = n_tp / (n_tp + n_sl) if (n_tp + n_sl) else 0.0 cbar = float(np.mean([(r["cost_px"] / r["risk"]) for r in deci])) if deci else 0.0 breakeven = (1.0 + cbar) / (1.0 + TARGET_R) # p*(1.5-c)=(1-p)*(1+c) w = [r["net_R"] for r in deci if r["exit_reason"] == "TP"] l = [-r["net_R"] for r in deci if r["exit_reason"] == "SL"] eff_r = (float(np.mean(w)) / float(np.mean(l))) if (w and l) else 0.0 eq = np.cumsum([r["net_R"] for r in sorted(rows, key=lambda r: (r["exit_time"], r["entry_time"]))]) maxdd = float((np.maximum.accumulate(eq) - eq).max()) if len(eq) else 0.0 is_rows = [r for r in deci if r["exit_time"] < SPLIT_MS] oos_rows = [r for r in deci if r["exit_time"] >= SPLIT_MS] both = sum(1 for r in rows if r["both_hit"]) return { "n_trades": len(rows), "n_tp": n_tp, "n_sl": n_sl, "n_end": n_end, "win_rate": round(win_rate, 4), "both_hit_trades": both, "target_R": TARGET_R, "spread_pips": cost["spread_pips"], "commission_usd_rt": round(COMM_USD_RT_PER_LOT * LOT, 4), "usdjpy_assumption": USDJPY_FIXED if symbol == "EURJPY" else "", "avg_cost_R": round(cbar, 4), "breakeven_win_rate": round(breakeven, 4), "effective_R": round(eff_r, 4), "maxdd_R": round(maxdd, 3), "is_n": len(is_rows), "is_winrate": round(sum(1 for r in is_rows if r["exit_reason"] == "TP") / len(is_rows), 4) if is_rows else 0.0, "oos_n": len(oos_rows), "oos_winrate": round(sum(1 for r in oos_rows if r["exit_reason"] == "TP") / len(oos_rows), 4) if oos_rows else 0.0, "period_start_ms": period_start_ms, "period_end_ms": period_end_ms, } def selftest_no_leak(strategy, df, p): """Genuine no-lookahead check: signals on a truncated prefix must be identical to the full-data signals on the same bars (causal indicators cannot change when later bars are removed).""" import pandas as pd K = 6000 sub = df.iloc[:K].copy().reset_index(drop=True) ind_full = compute_all(df, p) ind_sub = compute_all(sub, p) sig_full = {(i, d) for (i, d, _) in build_signals(strategy, df, ind_full, p) if i < K - 1} sig_sub = {(i, d) for (i, d, _) in build_signals(strategy, sub, ind_sub, p)} assert sig_full == sig_sub, "LEAK: truncation changed signals %d vs %d" % (len(sig_full), len(sig_sub)) return len(sig_sub) def run_portfolio(df, rows, symbol): """Trade engine receipt via vectorbtpro ``Portfolio.from_signals``. Two tiers, both executed in the engine (no import-only receipt): - Tier 1: every trade held at least one bar is replayed in a per-direction full-schedule portfolio at the ledger fills (entry/exit bars, entry/exit prices and per-order fees checked). - Tier 2: each zero-duration (same-bar entry+exit) trade is replayed in a single-trade 2-bar portfolio at its exact fills, because ``from_signals`` cannot open and close a position on the same bar. Raises unless every ledger trade reconciles. Returns version. """ import numpy as np import pandas as pd import vectorbtpro as vbt n = len(df) close = pd.Series(df["close"].to_numpy(), index=df.index, dtype=float) units = LOT * 100000.0 # 0.01 lot = 1000 units fee_side = COMM_USD_RT_PER_LOT * LOT / 2.0 # $0.035 per order normal = [r for r in rows if r["exit_bar"] > r["entry_bar"]] samebar = [r for r in rows if r["exit_bar"] == r["entry_bar"]] assert len(normal) + len(samebar) == len(rows), "exit before entry" for r in rows: assert r["entry_bar"] == r["sig_bar"] + 1, "entry must be next bar" checked = 0 total_fees = 0.0 for direction, vbt_dir in (("long", "longonly"), ("short", "shortonly")): sub = [r for r in normal if r["direction"] == direction] entries = np.zeros(n, dtype=bool) exits = np.zeros(n, dtype=bool) price = close.copy() for r in sub: assert not entries[r["entry_bar"]], "overlapping entries" entries[r["entry_bar"]] = True exits[r["exit_bar"]] = True price.iloc[r["entry_bar"]] = r["entry_fill"] price.iloc[r["exit_bar"]] = r["exit_fill"] assert not (entries & exits).any(), "entry/exit bar collision in tier 1" pf = vbt.Portfolio.from_signals( close=close, entries=entries, exits=exits, direction=vbt_dir, price=price, size=units, fixed_fees=fee_side, init_cash=10000.0, ) rec = pf.trades.records assert len(rec) == len(sub), "vbt count %d vs ledger %d (%s)" % (len(rec), len(sub), vbt_dir) want = {(r["entry_bar"], r["exit_bar"]): r for r in sub} for _, t in rec.iterrows(): key = (int(t["entry_idx"]), int(t["exit_idx"])) assert key in want, "unexpected vbt trade %r" % (key,) r = want.pop(key) assert abs(float(t["entry_price"]) - r["entry_fill"]) < 1e-9, key assert abs(float(t["exit_price"]) - r["exit_fill"]) < 1e-9, key assert not want, "ledger trades missing in vbt: %d" % len(want) total_fees += float(pf.orders.records["fees"].sum()) if len(pf.orders.records) else 0.0 checked += len(sub) assert abs(total_fees - 2.0 * fee_side * len(normal)) < 1e-6, total_fees for r in samebar: c2 = pd.Series([float(close.iloc[r["entry_bar"]])] * 2, dtype=float) p2 = pd.Series([float(r["entry_fill"]), float(r["exit_fill"])], dtype=float) pf = vbt.Portfolio.from_signals( close=c2, entries=np.array([True, False]), exits=np.array([False, True]), direction="longonly" if r["direction"] == "long" else "shortonly", price=p2, size=units, fixed_fees=fee_side, init_cash=10000.0, ) rec = pf.trades.records assert len(rec) == 1, "same-bar receipt failed" assert abs(float(rec["entry_price"].iloc[0]) - r["entry_fill"]) < 1e-9 assert abs(float(rec["exit_price"].iloc[0]) - r["exit_fill"]) < 1e-9 checked += 1 assert checked == len(rows), "vbt reconciliation failed: %d vs %d" % (checked, len(rows)) return str(getattr(vbt, "__version__", "unknown")) def selftest_receipt(): """Synthetic receipt check: one normal + one same-bar trade per direction. Maintained coverage for run_portfolio's two tiers. Runs under ``py -3.11 w7_vbt.py`` where vectorbtpro is installed. """ import numpy as np import pandas as pd n = 12 df = pd.DataFrame( { "timestamp": np.arange(n, dtype=np.int64) * 3600000, "open": np.linspace(1.10, 1.12, n), "high": np.linspace(1.101, 1.121, n), "low": np.linspace(1.099, 1.119, n), "close": np.linspace(1.10, 1.12, n), } ) rows = [ {"sig_bar": 1, "entry_bar": 2, "exit_bar": 5, "direction": "long", "entry_fill": 1.101, "exit_fill": 1.108}, {"sig_bar": 3, "entry_bar": 4, "exit_bar": 4, "direction": "long", "entry_fill": 1.103, "exit_fill": 1.102}, {"sig_bar": 5, "entry_bar": 6, "exit_bar": 9, "direction": "short", "entry_fill": 1.112, "exit_fill": 1.106}, {"sig_bar": 9, "entry_bar": 10, "exit_bar": 10, "direction": "short", "entry_fill": 1.118, "exit_fill": 1.119}, ] ver = run_portfolio(df, rows, "EURUSD") assert isinstance(ver, str) and ver, "no vbt version" return ver def ms_iso(ms): import datetime return datetime.datetime.fromtimestamp(ms / 1000, datetime.timezone(datetime.timedelta(hours=2))).isoformat() SUMMARY_COLS = [ "strategy", "variant", "scope", "period_start", "period_end", "split_iso", "bars", "n_trades", "n_tp", "n_sl", "n_end", "win_rate", "both_hit_trades", "skipped_degenerate", "skipped_wide", "target_R", "spread_pips", "commission_usd_rt", "usdjpy_assumption", "avg_cost_R", "breakeven_win_rate", "effective_R", "maxdd_R", "is_n", "is_winrate", "oos_n", "oos_winrate", "vbt_version", "vbt_reconciled", ] TRADE_COLS = [ "strategy", "variant", "symbol", "trade_no", "direction", "signal_time", "entry_time", "exit_time", "entry_fill", "exit_fill", "risk_px", "exit_reason", "both_hit", "cost_px", "net_R", "oos", ] def run_strategy(strategy, variants, outdir, use_vbt=True): """Full battery: base + variants x EURUSD/EURJPY. Writes summary/trades CSVs.""" import csv symbols = ["EURUSD", "EURJPY"] data = {s: load(s) for s in symbols} try: import vectorbtpro as vbt vbt_version = str(getattr(vbt, "__version__", "unknown")) except ImportError: if use_vbt: raise vbt_version = "none (logic validation only)" summary_rows = [] trade_rows = [] for v in variants: pooled = [] for symbol in symbols: df = data[symbol] ind = compute_all(df, v) selftest_no_leak(strategy, df, v) sigs = build_signals(strategy, df, ind, v) cost = cost_model(symbol) rows, sk_d, sk_w = walk(df, sigs, v, cost, ind["atr14"].to_numpy()) if use_vbt: ver = run_portfolio(df, rows, symbol) assert ver == vbt_version, (ver, vbt_version) s = summarize(rows, symbol, cost, int(df["timestamp"].iloc[0]), int(df["timestamp"].iloc[-1])) s.update( { "strategy": strategy, "variant": v["id"], "scope": symbol, "period_start": ms_iso(int(df["timestamp"].iloc[0])), "period_end": ms_iso(int(df["timestamp"].iloc[-1])), "split_iso": SPLIT_ISO, "bars": len(df), "skipped_degenerate": sk_d, "skipped_wide": sk_w, "vbt_version": vbt_version, "vbt_reconciled": "yes" if use_vbt else "no", } ) summary_rows.append(s) for k, r in enumerate(rows, 1): pooled.append((symbol, v["id"], k, r)) trade_rows.append( { "strategy": strategy, "variant": v["id"], "symbol": symbol, "trade_no": k, "direction": r["direction"], "signal_time": ms_iso(r["signal_time"]), "entry_time": ms_iso(r["entry_time"]), "exit_time": ms_iso(r["exit_time"]), "entry_fill": round(r["entry_fill"], 5), "exit_fill": round(r["exit_fill"], 5), "risk_px": round(r["risk"], 5), "exit_reason": r["exit_reason"], "both_hit": r["both_hit"], "cost_px": round(r["cost_px"], 5), "net_R": round(r["net_R"], 4), "oos": int(r["exit_time"] >= SPLIT_MS), } ) pall = [r for (_, _, _, r) in pooled] cost0 = cost_model("EURUSD") ps = summarize(pall, "EURUSD", cost0, int(data["EURUSD"]["timestamp"].iloc[0]), int(data["EURUSD"]["timestamp"].iloc[-1])) ps.update( { "strategy": strategy, "variant": v["id"], "scope": "POOLED", "period_start": ms_iso(int(data["EURUSD"]["timestamp"].iloc[0])), "period_end": ms_iso(int(data["EURUSD"]["timestamp"].iloc[-1])), "split_iso": SPLIT_ISO, "bars": len(data["EURUSD"]) + len(data["EURJPY"]), "skipped_degenerate": sum(s["skipped_degenerate"] for s in summary_rows if s["variant"] == v["id"]), "skipped_wide": sum(s["skipped_wide"] for s in summary_rows if s["variant"] == v["id"]), "spread_pips": "1.0/1.2", "usdjpy_assumption": USDJPY_FIXED, "vbt_version": vbt_version, "vbt_reconciled": "yes" if use_vbt else "no", } ) summary_rows.append(ps) os.makedirs(outdir, exist_ok=True) with open(os.path.join(outdir, "summary.csv"), "w", newline="") as f: w = csv.DictWriter(f, fieldnames=SUMMARY_COLS, extrasaction="ignore") w.writeheader() w.writerows(summary_rows) with open(os.path.join(outdir, "trades.csv"), "w", newline="") as f: w = csv.DictWriter(f, fieldnames=TRADE_COLS) w.writeheader() w.writerows(trade_rows) return summary_rows if __name__ == "__main__": print("vbt", selftest_receipt(), "receipt-ok")