448 lines
16 KiB
Python
448 lines
16 KiB
Python
"""CSV-fixture tests for scripts.stats — compute_stats, render_stats,
|
|
compute_calibration, render_calibration, main()."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import csv
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
|
|
|
from scripts.append_row import CSV_COLUMNS # noqa: E402
|
|
from scripts.stats import ( # noqa: E402
|
|
CORE_CALIBRATION_FIELDS,
|
|
compute_calibration,
|
|
compute_stats,
|
|
main,
|
|
render_calibration,
|
|
render_stats,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Fixture row builder
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _base_row(**overrides) -> dict[str, str]:
|
|
base = {
|
|
"id": "0",
|
|
"screenshot_file": "",
|
|
"source": "vision",
|
|
"data": "2026-05-13",
|
|
"zi": "Mi",
|
|
"ora_ro": "17:30",
|
|
"ora_utc": "14:30",
|
|
"instrument": "DIA",
|
|
"directie": "Buy",
|
|
"tf_mare": "5min",
|
|
"tf_mic": "1min",
|
|
"calitate": "Clară",
|
|
"entry": "400.0",
|
|
"sl": "399.0",
|
|
"tp0": "400.5",
|
|
"tp1": "401.0",
|
|
"tp2": "402.0",
|
|
"risc_pct": "0.25",
|
|
"outcome_path": "TP0→TP1",
|
|
"max_reached": "TP1",
|
|
"be_moved": "True",
|
|
"pl_marius": "0.5000",
|
|
"pl_theoretical": "0.3330",
|
|
"set": "A2",
|
|
"indicator_version": "v-2026-05",
|
|
"pl_overlay_version": "marius-v1",
|
|
"csv_schema_version": "1",
|
|
"extracted_at": "2026-05-13T10:00:00Z",
|
|
"note": "",
|
|
}
|
|
base.update({k: str(v) for k, v in overrides.items()})
|
|
return base
|
|
|
|
|
|
def _write_csv(path: Path, rows: list[dict[str, str]]) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
with path.open("w", encoding="utf-8", newline="") as fh:
|
|
w = csv.DictWriter(fh, fieldnames=list(CSV_COLUMNS))
|
|
w.writeheader()
|
|
for r in rows:
|
|
w.writerow({k: r.get(k, "") for k in CSV_COLUMNS})
|
|
|
|
|
|
# Outcome templates (P/L values) — match scripts.pl_calc tables.
|
|
_SL = {"outcome_path": "SL", "max_reached": "SL_first", "be_moved": "False",
|
|
"pl_marius": "-1.0000", "pl_theoretical": "-1.0000"}
|
|
_TP0_SL_BE = {"outcome_path": "TP0→SL", "max_reached": "TP0", "be_moved": "True",
|
|
"pl_marius": "0.2000", "pl_theoretical": "0.1330"}
|
|
_TP0_TP1 = {"outcome_path": "TP0→TP1", "max_reached": "TP1", "be_moved": "True",
|
|
"pl_marius": "0.5000", "pl_theoretical": "0.3330"}
|
|
_TP0_TP2 = {"outcome_path": "TP0→TP2", "max_reached": "TP2", "be_moved": "True",
|
|
"pl_marius": "0.5000", "pl_theoretical": "0.6670"}
|
|
_PENDING = {"outcome_path": "pending", "max_reached": "TP0", "be_moved": "False",
|
|
"pl_marius": "", "pl_theoretical": "0.1330"}
|
|
|
|
|
|
def _synthetic_csv(tmp_path: Path) -> Path:
|
|
"""30-trade backtest fixture.
|
|
|
|
Set distribution:
|
|
A1: 8 rows (all closed; 3 SL, 2 TP0→SL, 2 TP0→TP1, 1 TP0→TP2)
|
|
A2: 10 rows (all closed; 4 SL, 3 TP0→SL, 2 TP0→TP1, 1 TP0→TP2)
|
|
B : 7 rows (2 pending, 5 closed; 2 SL, 2 TP0→TP1, 1 TP0→TP2)
|
|
D : 5 rows (3 pending, 2 closed; 1 SL, 1 TP0→TP1)
|
|
|
|
Totals: n_total=30, n_pending=5, n_closed=25.
|
|
|
|
Wins by pl_marius (>0): all TP0→SL_BE + TP0→TP1 + TP0→TP2
|
|
A1: 2 + 2 + 1 = 5 wins / 8
|
|
A2: 3 + 2 + 1 = 6 wins / 10
|
|
B : 0 + 2 + 1 = 3 wins / 5
|
|
D : 0 + 1 + 0 = 1 win / 2
|
|
Total wins = 15 / 25 = 60.0%.
|
|
|
|
Calitate distribution: half "Clară", half "Slabă" (alternating).
|
|
Directie distribution: 2/3 Buy, 1/3 Sell.
|
|
"""
|
|
rows: list[dict[str, str]] = []
|
|
rid = 0
|
|
|
|
def add(set_label: str, outcomes: list[dict[str, str]]) -> None:
|
|
nonlocal rid
|
|
for i, outcome in enumerate(outcomes):
|
|
rid += 1
|
|
row = _base_row(
|
|
id=rid,
|
|
screenshot_file=f"{set_label.lower()}-{rid}.png",
|
|
set=set_label,
|
|
calitate="Clară" if rid % 2 == 0 else "Slabă",
|
|
directie="Buy" if rid % 3 != 0 else "Sell",
|
|
)
|
|
row.update({k: str(v) for k, v in outcome.items()})
|
|
rows.append(row)
|
|
|
|
add("A1", [_SL] * 3 + [_TP0_SL_BE] * 2 + [_TP0_TP1] * 2 + [_TP0_TP2] * 1)
|
|
add("A2", [_SL] * 4 + [_TP0_SL_BE] * 3 + [_TP0_TP1] * 2 + [_TP0_TP2] * 1)
|
|
add("B", [_PENDING] * 2 + [_SL] * 2 + [_TP0_TP1] * 2 + [_TP0_TP2] * 1)
|
|
add("D", [_PENDING] * 3 + [_SL] * 1 + [_TP0_TP1] * 1)
|
|
|
|
path = tmp_path / "jurnal.csv"
|
|
_write_csv(path, rows)
|
|
return path
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# compute_stats — core
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestComputeStats:
|
|
def test_compute_stats_n_pending(self, tmp_path: Path) -> None:
|
|
path = _synthetic_csv(tmp_path)
|
|
s = compute_stats(path)
|
|
assert s["n_total"] == 30
|
|
assert s["n_pending"] == 5
|
|
assert s["n_closed"] == 25
|
|
|
|
def test_compute_stats_wr_correct(self, tmp_path: Path) -> None:
|
|
"""Manual win count: 15 / 25 = 60.0%."""
|
|
path = _synthetic_csv(tmp_path)
|
|
s = compute_stats(path)
|
|
assert s["wr"] == pytest.approx(15 / 25)
|
|
lo, hi = s["wr_ci_95"]
|
|
assert 0.0 <= lo <= s["wr"] <= hi <= 1.0
|
|
|
|
def test_compute_stats_per_set(self, tmp_path: Path) -> None:
|
|
path = _synthetic_csv(tmp_path)
|
|
s = compute_stats(path)
|
|
a2 = s["per_set"]["A2"]
|
|
assert a2["n"] == 10 # 10 closed A2 trades
|
|
# A2 wins (pl_marius > 0): 3 BE + 2 TP1 + 1 TP2 = 6 / 10
|
|
assert a2["wr"] == pytest.approx(0.60)
|
|
|
|
def test_per_set_b_pending_excluded(self, tmp_path: Path) -> None:
|
|
"""Set B has 7 total rows (2 pending + 5 closed). n must be 5."""
|
|
path = _synthetic_csv(tmp_path)
|
|
s = compute_stats(path)
|
|
assert s["per_set"]["B"]["n"] == 5
|
|
# B wins: 0 BE + 2 TP1 + 1 TP2 = 3 / 5
|
|
assert s["per_set"]["B"]["wr"] == pytest.approx(0.60)
|
|
|
|
def test_per_directie_no_ci_keys(self, tmp_path: Path) -> None:
|
|
"""per_directie omits CI fields per spec (only n / wr / expectancy)."""
|
|
path = _synthetic_csv(tmp_path)
|
|
s = compute_stats(path)
|
|
for k, d in s["per_directie"].items():
|
|
assert set(d.keys()) == {"n", "wr", "expectancy"}, k
|
|
|
|
def test_overlay_theoretical_vs_marius(self, tmp_path: Path) -> None:
|
|
path = _synthetic_csv(tmp_path)
|
|
s_m = compute_stats(path, overlay="pl_marius")
|
|
s_t = compute_stats(path, overlay="pl_theoretical")
|
|
# Same N, but different expectancy.
|
|
assert s_m["n_closed"] == s_t["n_closed"]
|
|
assert s_m["expectancy"] != s_t["expectancy"]
|
|
|
|
def test_unknown_overlay_raises(self, tmp_path: Path) -> None:
|
|
path = _synthetic_csv(tmp_path)
|
|
with pytest.raises(ValueError):
|
|
compute_stats(path, overlay="pl_imaginary")
|
|
|
|
def test_empty_csv_no_crash(self, tmp_path: Path) -> None:
|
|
path = tmp_path / "empty.csv"
|
|
_write_csv(path, [])
|
|
s = compute_stats(path)
|
|
assert s["n_total"] == 0
|
|
assert s["n_closed"] == 0
|
|
assert s["per_set"] == {}
|
|
assert s["wr"] == 0.0
|
|
assert s["wr_ci_95"] == (0.0, 0.0)
|
|
|
|
def test_missing_csv_no_crash(self, tmp_path: Path) -> None:
|
|
# Nonexistent path: treat as empty, do not raise.
|
|
s = compute_stats(tmp_path / "ghost.csv")
|
|
assert s["n_total"] == 0
|
|
|
|
def test_calibration_rows_excluded(self, tmp_path: Path) -> None:
|
|
rows = [
|
|
_base_row(id=1, source="vision", screenshot_file="v.png"),
|
|
_base_row(id=2, source="manual_calibration", screenshot_file="c.png"),
|
|
_base_row(id=3, source="vision_calibration", screenshot_file="c.png"),
|
|
]
|
|
path = tmp_path / "j.csv"
|
|
_write_csv(path, rows)
|
|
s = compute_stats(path)
|
|
assert s["n_total"] == 1 # calibration rows filtered out
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# render_stats
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestRenderStats:
|
|
def test_render_stats_no_crash(self, tmp_path: Path) -> None:
|
|
path = _synthetic_csv(tmp_path)
|
|
s = compute_stats(path)
|
|
out = render_stats(s, "pl_marius")
|
|
assert isinstance(out, str)
|
|
assert out # non-empty
|
|
assert "STOPPING RULE" in out
|
|
|
|
def test_render_stats_contains_sections(self, tmp_path: Path) -> None:
|
|
path = _synthetic_csv(tmp_path)
|
|
out = render_stats(compute_stats(path), "pl_marius")
|
|
for marker in (
|
|
"Stats jurnal.csv",
|
|
"Trade-uri totale",
|
|
"GLOBAL",
|
|
"PER SET:",
|
|
"PER CALITATE",
|
|
"PER DIRECȚIE",
|
|
"DESCRIPTOR ONLY",
|
|
):
|
|
assert marker in out, f"missing section: {marker!r}"
|
|
|
|
def test_render_stats_flags_under_threshold(self, tmp_path: Path) -> None:
|
|
"""All Sets in synthetic fixture have N<40 → all should be flagged."""
|
|
path = _synthetic_csv(tmp_path)
|
|
out = render_stats(compute_stats(path), "pl_marius")
|
|
for k in ("A1", "A2", "B", "D"):
|
|
assert f"{k}: N=" in out
|
|
assert "NEEDS MORE DATA" in out
|
|
|
|
def test_render_stats_empty(self, tmp_path: Path) -> None:
|
|
path = tmp_path / "empty.csv"
|
|
_write_csv(path, [])
|
|
out = render_stats(compute_stats(path), "pl_marius")
|
|
assert "Trade-uri totale: 0" in out
|
|
# No crash, no per-Set table for an empty dataset.
|
|
assert "NEEDS MORE DATA" not in out
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# compute_calibration
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestComputeCalibration:
|
|
def test_compute_calibration_pairs(self, tmp_path: Path) -> None:
|
|
rows: list[dict[str, str]] = []
|
|
for i in range(5):
|
|
f = f"cal-{i}.png"
|
|
rows.append(_base_row(
|
|
id=i * 2 + 1, source="manual_calibration", screenshot_file=f
|
|
))
|
|
rows.append(_base_row(
|
|
id=i * 2 + 2, source="vision_calibration", screenshot_file=f
|
|
))
|
|
path = tmp_path / "j.csv"
|
|
_write_csv(path, rows)
|
|
cal = compute_calibration(path)
|
|
assert cal["n_pairs"] == 5
|
|
for fld in CORE_CALIBRATION_FIELDS:
|
|
assert fld in cal["fields"]
|
|
# All identical → 5 matches, 0 mismatches per field.
|
|
assert cal["fields"][fld]["match"] == 5
|
|
assert cal["fields"][fld]["mismatch"] == 0
|
|
assert cal["fields"][fld]["match_rate"] == pytest.approx(1.0)
|
|
|
|
def test_compute_calibration_mismatch_examples(self, tmp_path: Path) -> None:
|
|
"""Modify entry on 2 pairs → mismatch_examples contains both."""
|
|
rows: list[dict[str, str]] = []
|
|
for i in range(5):
|
|
f = f"cal-{i}.png"
|
|
manual_entry = "400.0"
|
|
# First two pairs differ on entry; the rest match exactly.
|
|
vision_entry = "401.5" if i < 2 else "400.0"
|
|
rows.append(_base_row(
|
|
id=i * 2 + 1, source="manual_calibration",
|
|
screenshot_file=f, entry=manual_entry,
|
|
))
|
|
rows.append(_base_row(
|
|
id=i * 2 + 2, source="vision_calibration",
|
|
screenshot_file=f, entry=vision_entry,
|
|
))
|
|
path = tmp_path / "j.csv"
|
|
_write_csv(path, rows)
|
|
cal = compute_calibration(path)
|
|
assert cal["n_pairs"] == 5
|
|
entry = cal["fields"]["entry"]
|
|
assert entry["match"] == 3
|
|
assert entry["mismatch"] == 2
|
|
assert entry["match_rate"] == pytest.approx(3 / 5)
|
|
assert len(entry["mismatch_examples"]) == 2
|
|
for ex in entry["mismatch_examples"]:
|
|
assert "manual=" in ex and "vision=" in ex
|
|
|
|
def test_calibration_examples_capped_at_3(self, tmp_path: Path) -> None:
|
|
"""5 mismatches but mismatch_examples is capped at 3."""
|
|
rows: list[dict[str, str]] = []
|
|
for i in range(5):
|
|
f = f"cal-{i}.png"
|
|
rows.append(_base_row(
|
|
id=i * 2 + 1, source="manual_calibration",
|
|
screenshot_file=f, entry="400.0",
|
|
))
|
|
rows.append(_base_row(
|
|
id=i * 2 + 2, source="vision_calibration",
|
|
screenshot_file=f, entry="500.0",
|
|
))
|
|
path = tmp_path / "j.csv"
|
|
_write_csv(path, rows)
|
|
cal = compute_calibration(path)
|
|
assert cal["fields"]["entry"]["mismatch"] == 5
|
|
assert len(cal["fields"]["entry"]["mismatch_examples"]) == 3
|
|
|
|
def test_calibration_numeric_tolerance(self, tmp_path: Path) -> None:
|
|
"""Floats within 0.01 must NOT count as a mismatch."""
|
|
rows = [
|
|
_base_row(
|
|
id=1, source="manual_calibration",
|
|
screenshot_file="cal-1.png", entry="400.005",
|
|
),
|
|
_base_row(
|
|
id=2, source="vision_calibration",
|
|
screenshot_file="cal-1.png", entry="400.010",
|
|
),
|
|
]
|
|
path = tmp_path / "j.csv"
|
|
_write_csv(path, rows)
|
|
cal = compute_calibration(path)
|
|
assert cal["fields"]["entry"]["match"] == 1
|
|
assert cal["fields"]["entry"]["mismatch"] == 0
|
|
|
|
def test_calibration_outside_tolerance(self, tmp_path: Path) -> None:
|
|
"""Floats > 0.01 apart DO count as a mismatch."""
|
|
rows = [
|
|
_base_row(
|
|
id=1, source="manual_calibration",
|
|
screenshot_file="cal-1.png", entry="400.00",
|
|
),
|
|
_base_row(
|
|
id=2, source="vision_calibration",
|
|
screenshot_file="cal-1.png", entry="400.05",
|
|
),
|
|
]
|
|
path = tmp_path / "j.csv"
|
|
_write_csv(path, rows)
|
|
cal = compute_calibration(path)
|
|
assert cal["fields"]["entry"]["mismatch"] == 1
|
|
|
|
def test_calibration_no_pairs(self, tmp_path: Path) -> None:
|
|
"""No paired screenshot → n_pairs=0, all rates 0.0."""
|
|
path = tmp_path / "j.csv"
|
|
_write_csv(path, [
|
|
_base_row(id=1, source="manual_calibration", screenshot_file="lonely.png"),
|
|
])
|
|
cal = compute_calibration(path)
|
|
assert cal["n_pairs"] == 0
|
|
for fld in CORE_CALIBRATION_FIELDS:
|
|
assert cal["fields"][fld]["match"] == 0
|
|
assert cal["fields"][fld]["mismatch"] == 0
|
|
|
|
def test_render_calibration_no_crash(self, tmp_path: Path) -> None:
|
|
rows = [
|
|
_base_row(id=1, source="manual_calibration",
|
|
screenshot_file="cal-1.png", directie="Buy"),
|
|
_base_row(id=2, source="vision_calibration",
|
|
screenshot_file="cal-1.png", directie="Sell",
|
|
entry="400.0", sl="401.0", tp0="399.5",
|
|
tp1="399.0", tp2="398.0"),
|
|
]
|
|
path = tmp_path / "j.csv"
|
|
_write_csv(path, rows)
|
|
out = render_calibration(compute_calibration(path))
|
|
assert "Calibration P4" in out
|
|
assert "directie" in out
|
|
|
|
def test_render_calibration_empty(self, tmp_path: Path) -> None:
|
|
path = tmp_path / "empty.csv"
|
|
_write_csv(path, [])
|
|
out = render_calibration(compute_calibration(path))
|
|
assert "0" in out
|
|
assert "FAIL" not in out
|
|
assert "PASS" not in out
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CLI
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestCLI:
|
|
def test_main_stats(
|
|
self, tmp_path: Path, capsys: pytest.CaptureFixture
|
|
) -> None:
|
|
path = _synthetic_csv(tmp_path)
|
|
rc = main(["--csv", str(path)])
|
|
assert rc == 0
|
|
assert "Stats jurnal.csv" in capsys.readouterr().out
|
|
|
|
def test_main_overlay(
|
|
self, tmp_path: Path, capsys: pytest.CaptureFixture
|
|
) -> None:
|
|
path = _synthetic_csv(tmp_path)
|
|
rc = main(["--csv", str(path), "--overlay", "pl_theoretical"])
|
|
assert rc == 0
|
|
assert "pl_theoretical" in capsys.readouterr().out
|
|
|
|
def test_main_calibration(
|
|
self, tmp_path: Path, capsys: pytest.CaptureFixture
|
|
) -> None:
|
|
rows = [
|
|
_base_row(id=1, source="manual_calibration",
|
|
screenshot_file="cal-1.png"),
|
|
_base_row(id=2, source="vision_calibration",
|
|
screenshot_file="cal-1.png"),
|
|
]
|
|
path = tmp_path / "j.csv"
|
|
_write_csv(path, rows)
|
|
rc = main(["--csv", str(path), "--calibration"])
|
|
assert rc == 0
|
|
out = capsys.readouterr().out
|
|
assert "Calibration P4" in out
|
|
assert "PASS" in out
|