import os, tempfile def _fresh_db(monkeypatch): tmp = tempfile.mkdtemp() path = os.path.join(tmp, "lotto.db") from app import db monkeypatch.setattr(db, "DB_PATH", path) db.init_db() return db def test_backtest_tables_exist(monkeypatch): db = _fresh_db(monkeypatch) with db._conn() as conn: tables = {r["name"] for r in conn.execute( "SELECT name FROM sqlite_master WHERE type='table'").fetchall()} assert "backtest_runs" in tables assert "winner_calibration" in tables def test_backtest_runs_unique(monkeypatch): db = _fresh_db(monkeypatch) db.save_backtest_run(draw_no=100, strategy="random_null", weight_label="-", weight_json=None, trial_id=None, n_tickets=10, hist={"m3":1,"m4":0,"m5":0,"m6":0,"bonus_hits":0}, best_match=3, avg_meta_score=0.5) db.save_backtest_run(draw_no=100, strategy="random_null", weight_label="-", weight_json=None, trial_id=None, n_tickets=10, hist={"m3":2,"m4":0,"m5":0,"m6":0,"bonus_hits":0}, best_match=3, avg_meta_score=0.6) # 멱등 upsert rows = db.get_backtest_runs(draw_no=100) assert len(rows) == 1 assert rows[0]["m3"] == 2 # 마지막 값으로 갱신 _SCORES = { "score_total": 1.23, "score_frequency": 0.30, "score_fingerprint": 0.25, "score_gap": 0.20, "score_cooccur": 0.28, "score_diversity": 0.20, } def test_winner_calibration_upsert(monkeypatch): """save_winner_calibration 두 번 호출 시 upsert — 행 1개, 값은 마지막 것.""" db = _fresh_db(monkeypatch) winning = [3, 7, 15, 22, 33, 41] db.save_winner_calibration(draw_no=200, winning=winning, scores=_SCORES, percentile=75.0, my_pick_avg=0.9, cache_draws=100) # 두 번째 저장 — percentile, my_pick_avg 업데이트 scores2 = {**_SCORES, "score_total": 2.00} db.save_winner_calibration(draw_no=200, winning=winning, scores=scores2, percentile=80.0, my_pick_avg=1.1, cache_draws=110) row = db.get_winner_calibration(200) assert row is not None # 행이 1개만 존재하는지 확인 with db._conn() as conn: cnt = conn.execute( "SELECT COUNT(*) AS c FROM winner_calibration WHERE draw_no=200" ).fetchone()["c"] assert cnt == 1 assert row["percentile"] == 80.0 assert row["score_total"] == 2.00 def _seed_draws(db, n=40): rows = [] import random as _r; _r.seed(2) for i in range(1, n + 1): s = sorted(_r.sample(range(1, 46), 6)) rows.append({"drw_no": i, "drw_date": f"2020-01-{(i%28)+1:02d}", "n1": s[0], "n2": s[1], "n3": s[2], "n4": s[3], "n5": s[4], "n6": s[5], "bonus": ((s[5] % 45) + 1)}) db.upsert_many_draws(rows) def test_backfill_calibration_idempotent(monkeypatch): db = _fresh_db(monkeypatch) _seed_draws(db, 40) from app import backtest as bt r1 = bt.backfill_calibration(batch=15, sample_m=200) # 첫 회차는 point-in-time 데이터가 빈약 → min_history 이후만 처리 done1 = len(db.get_calibrated_draw_nos()) assert done1 > 0 r2 = bt.backfill_calibration(batch=100, sample_m=200) # 나머지 done2 = len(db.get_calibrated_draw_nos()) assert done2 >= done1 r3 = bt.backfill_calibration(batch=100, sample_m=200) # 재실행 → 추가 0 assert r3["calibrated"] == 0 def test_run_forward_purchase_persists_all_strategies(monkeypatch): db = _fresh_db(monkeypatch) _seed_draws(db, 40) from app import backtest as bt # 작은 규모로 빠르게 res = bt.run_forward_purchase(draw_no=40, k=20, pool_n=500, sample_seed=5) assert res["ok"] is True rows = db.get_backtest_runs(draw_no=40) strategies = {r["strategy"] for r in rows} assert "random_null" in strategies assert "coverage" in strategies assert "engine_w" in strategies # base 가중치로 최소 1건 for r in rows: assert r["n_tickets"] == 20 def test_calibrate_winner_no_draw(monkeypatch): """DB에 없는 회차 번호 → ok=False, reason='no_draw'.""" db = _fresh_db(monkeypatch) _seed_draws(db, 40) from app import backtest as bt r = bt.calibrate_winner(99999) assert r["ok"] is False assert r["reason"] == "no_draw" def test_calibrate_winner_insufficient_history(monkeypatch): """point-in-time 이력이 MIN_HISTORY(30) 미만인 회차 → reason='insufficient_history'. draw_no=20이면 PIT 이력이 19개(draws 1~19)로 30 미만.""" db = _fresh_db(monkeypatch) _seed_draws(db, 40) from app import backtest as bt r = bt.calibrate_winner(20) assert r["ok"] is False assert r["reason"] == "insufficient_history" def test_run_forward_purchase_with_trials(monkeypatch): """그 주 weight_trials가 존재하면 engine_w 행의 weight_label이 'w0'..'w5' 형식이어야 한다.""" db = _fresh_db(monkeypatch) _seed_draws(db, 40) # draw 40: drw_date='2020-01-13' → week_start='2020-01-13' from datetime import date, timedelta draw_date = date.fromisoformat("2020-01-13") ws = (draw_date - timedelta(days=draw_date.weekday())).isoformat() # 해당 주에 trial 2개 심기 (day_of_week 0, 1) db.save_weight_trial(ws, 0, [0.1, 0.3, 0.2, 0.2, 0.2], "perturb") db.save_weight_trial(ws, 1, [0.25, 0.25, 0.25, 0.15, 0.1], "perturb") from app import backtest as bt res = bt.run_forward_purchase(draw_no=40, k=20, pool_n=500, sample_seed=5) assert res["ok"] is True rows = db.get_backtest_runs(draw_no=40) engine_w_labels = {r["weight_label"] for r in rows if r["strategy"] == "engine_w"} # trials가 있으므로 'base'가 아닌 'w0', 'w1' 형식이어야 한다 assert "base" not in engine_w_labels assert any(lbl.startswith("w") for lbl in engine_w_labels) def test_run_forward_purchase_idempotent(monkeypatch): """run_forward_purchase 두 번 호출 시 upsert — 행 수 변화 없음.""" db = _fresh_db(monkeypatch) _seed_draws(db, 40) from app import backtest as bt bt.run_forward_purchase(draw_no=40, k=20, pool_n=500, sample_seed=5) count_after_first = len(db.get_backtest_runs(draw_no=40)) bt.run_forward_purchase(draw_no=40, k=20, pool_n=500, sample_seed=5) count_after_second = len(db.get_backtest_runs(draw_no=40)) assert count_after_second == count_after_first def test_get_calibrated_draw_nos(monkeypatch): """저장된 draw_no 집합이 get_calibrated_draw_nos에 포함되어야 한다.""" db = _fresh_db(monkeypatch) winning = [1, 2, 3, 4, 5, 6] for draw_no in (301, 302, 303): db.save_winner_calibration(draw_no=draw_no, winning=winning, scores=_SCORES, percentile=50.0, my_pick_avg=0.5, cache_draws=50) nos = db.get_calibrated_draw_nos() assert isinstance(nos, set) assert {301, 302, 303}.issubset(nos)