from datetime import datetime, timedelta, timezone from app.config import MODEL_VERSION from app.core.calibration import calibrate_probability from app.core.competitions import competition_for_sport_key, season_start_year from app.core.history import append_new_picks, performance_metrics, settle_history from app.core.market import market_consensus, remove_vig from app.core.names import build_team_catalog, resolve_event_pair from app.core.stats import ( build_elo, dixon_coles_1x2, poisson_1x2, predictive_models, ) from app.core.tickets import optimize_ticket from app.models import FinishedMatch from app.providers.football_data import FootballDataProvider NOW = datetime.now(timezone.utc) def sample_matches(competition="PL", n=120): out = [] scores = [ ("Alpha FC", "Gamma FC", 2, 0), ("Beta FC", "Delta FC", 1, 1), ("Alpha FC", "Delta FC", 3, 1), ("Gamma FC", "Beta FC", 0, 2), ("Gamma FC", "Alpha FC", 1, 2), ("Delta FC", "Beta FC", 0, 1), ("Delta FC", "Alpha FC", 0, 2), ("Beta FC", "Gamma FC", 2, 0), ] ids = {"Alpha FC": "1", "Beta FC": "2", "Gamma FC": "3", "Delta FC": "4"} for i in range(n): h, a, hg, ag = scores[i % len(scores)] out.append(FinishedMatch( match_id=str(i), competition=competition, utc_date=NOW - timedelta(days=2 * (n - i)), home=h, away=a, home_goals=hg, away_goals=ag, home_id=ids[h], away_id=ids[a], home_aliases=(h, h.replace(" FC", ""), ids[h]), away_aliases=(a, a.replace(" FC", ""), ids[a]), )) return out def make_event(): books = [] prices = [ (1.50, 4.20, 7.20), (1.52, 4.10, 7.00), (1.48, 4.30, 7.40), (1.51, 4.15, 7.10), ] for i, (ho, do, ao) in enumerate(prices): books.append({ "key": f"b{i}", "title": f"Book {i}", "last_update": NOW.isoformat(), "markets": [{ "key": "h2h", "outcomes": [ {"name": "Alpha FC", "price": ho}, {"name": "Draw", "price": do}, {"name": "Beta FC", "price": ao}, ], }], }) return { "id": "evt1", "sport_key": "soccer_epl", "_sport_key": "soccer_epl", "home_team": "Alpha FC", "away_team": "Beta FC", "commence_time": (NOW + timedelta(hours=8)).isoformat(), "bookmakers": books, } def test_competition_mapping(): spec = competition_for_sport_key("soccer_epl") assert spec is not None assert spec.football_data_code == "PL" assert season_start_year(spec, NOW.date()) in {NOW.year, NOW.year - 1} def test_remove_vig_sums_to_one(): p = remove_vig(1.60, 4.00, 6.00) assert abs(sum(p) - 1.0) < 1e-9 assert all(0 < x < 1 for x in p) def test_market_consensus_devigs_each_book_and_has_depth(): market = market_consensus(make_event()) assert market.bookmakers == 4 assert market.home_prob is not None assert abs(market.home_prob + market.draw_prob + market.away_prob - 1.0) < 1e-9 assert market.dispersion < 0.03 assert 1.45 < market.home_odd < 1.55 def test_name_resolution_is_competition_scoped(): matches = sample_matches("PL", 80) catalog = build_team_catalog(matches, "PL") home, away, confidence, detail = resolve_event_pair("Alpha", "Beta", catalog) assert home is not None and away is not None assert home.name == "Alpha FC" assert away.name == "Beta FC" assert confidence > 0.95 def test_poisson_sums_to_one(): p = poisson_1x2(1.8, 0.9) assert abs(sum(p) - 1.0) < 1e-8 assert p[0] > p[2] def test_dixon_coles_sums_to_one_and_changes_draw(): independent = dixon_coles_1x2(1.35, 1.05, 0.0) corrected = dixon_coles_1x2(1.35, 1.05, -0.08) assert abs(sum(corrected) - 1.0) < 1e-8 assert corrected[1] != independent[1] def test_predictive_models_are_valid_and_detailed(): matches = sample_matches() elo = build_elo(matches) result = predictive_models( "id:1", "id:2", matches, elo, competition="PL", as_of=NOW + timedelta(hours=1), ) for key in ("poisson", "elo", "form", "ensemble"): probs = result[key] assert abs(sum(probs) - 1.0) < 1e-8 assert all(0 <= x <= 1 for x in probs) assert 0 <= result["quality"] <= 1 assert result["league_sample"] >= 60 assert result["lambda_home"] > 0 assert result["lambda_away"] > 0 def test_football_data_uses_regular_time_for_knockout(): item = { "id": 99, "utcDate": NOW.isoformat(), "competition": {"code": "CL"}, "homeTeam": {"id": 1, "name": "Home FC", "shortName": "Home", "tla": "HOM"}, "awayTeam": {"id": 2, "name": "Away FC", "shortName": "Away", "tla": "AWY"}, "score": { "regularTime": {"home": 1, "away": 1}, "fullTime": {"home": 2, "away": 1}, }, } match = FootballDataProvider._parse_match(item) assert match is not None assert (match.home_goals, match.away_goals) == (1, 1) def test_calibration_is_inactive_on_tiny_sample(): history = [{ "result": "win", "probability": 0.70, "model_version": MODEL_VERSION, "competition_code": "PL", }] * 5 p, meta = calibrate_probability( 0.70, history, model_version=MODEL_VERSION, competition_code="PL" ) assert p == 0.70 assert meta["delta"] == 0.0 def test_calibration_adjusts_after_enough_forward_evidence(): history = [] for i in range(30): history.append({ "result": "win" if i < 26 else "loss", "probability": 0.70, "model_version": MODEL_VERSION, "competition_code": "PL", }) p, meta = calibrate_probability( 0.70, history, model_version=MODEL_VERSION, competition_code="PL" ) assert meta["effective_samples"] >= 12 assert p > 0.70 assert p <= 0.75 def test_append_history_does_not_duplicate_same_event_with_new_side(): history = [] base = { "event_id": "e1", "kickoff": NOW.isoformat(), "competition": "Premier League", "competition_code": "PL", "home": "Alpha", "away": "Beta", "selection": "Alpha", "side": "home", "odd": 1.5, "probability": .7, "safe_score": 85, "model_version": MODEL_VERSION, } append_new_picks(history, [base]) flipped = dict(base) flipped.update(selection="Beta", side="away") append_new_picks(history, [flipped]) assert len(history) == 1 def test_settlement_respects_competition_and_pair(): finished_at = NOW - timedelta(hours=3) matches = [ FinishedMatch( match_id="1", competition="PL", utc_date=finished_at, home="Alpha FC", away="Beta FC", home_goals=2, away_goals=0, home_aliases=("Alpha",), away_aliases=("Beta",), ), FinishedMatch( match_id="2", competition="SA", utc_date=finished_at, home="Alpha FC", away="Beta FC", home_goals=0, away_goals=2, home_aliases=("Alpha",), away_aliases=("Beta",), ), ] history = [{ "event_id": "e", "kickoff": finished_at.isoformat(), "competition_code": "PL", "home": "Alpha", "away": "Beta", "side": "home", "odd": 1.5, "probability": .7, "result": None, }] settle_history(history, matches) assert history[0]["result"] == "win" def test_performance_metrics_have_calibration_and_drawdown(): history = [ {"result": "win", "probability": .70, "profit_units": .5}, {"result": "loss", "probability": .70, "profit_units": -1}, {"result": "win", "probability": .65, "profit_units": .6}, ] metrics = performance_metrics(history) assert metrics["settled"] == 3 assert metrics["brier_score"] is not None assert metrics["log_loss"] is not None assert metrics["max_drawdown_units"] >= 0 def test_ticket_optimizer_respects_max_legs_and_stresses_dependency(): picks = [] for i, odd in enumerate([1.4, 1.5, 1.6, 1.7, 1.8]): picks.append({ "event_id": str(i), "home": f"H{i}", "away": f"A{i}", "selection": f"H{i}", "odd": odd, "probability": .72, "conservative_probability": .63, "safe_score": 85, "competition_code": "PL" if i < 3 else "SA", "kickoff": (NOW + timedelta(hours=i)).isoformat(), }) ticket = optimize_ticket(picks, target_odd=4.0, max_legs=4) assert ticket is not None assert 2 <= len(ticket["legs"]) <= 4 assert ticket["total_odd"] > 1 assert ticket["dependency_factor"] <= 1.0 assert isinstance(ticket["target_met"], bool) def test_ticket_target_requires_reaching_the_full_advertised_odd(): picks = [ { "event_id": "1", "home": "A", "away": "B", "selection": "A", "odd": 1.4, "probability": .72, "conservative_probability": .63, "safe_score": 85, "competition_code": "PL", "kickoff": NOW.isoformat(), }, { "event_id": "2", "home": "C", "away": "D", "selection": "C", "odd": 1.75, "probability": .72, "conservative_probability": .63, "safe_score": 85, "competition_code": "SA", "kickoff": NOW.isoformat(), }, ] ticket = optimize_ticket(picks, target_odd=2.5, max_legs=2) assert ticket is not None assert ticket["total_odd"] == 2.45 assert ticket["target_met"] is False def _strong_favorite_fixture(): matches = [] ids = {"Alpha FC": "1", "Beta FC": "2", "Gamma FC": "3", "Delta FC": "4"} for i in range(160): dt = NOW - timedelta(days=1 + (160 - i)) if i % 4 == 0: h, a, hg, ag = "Alpha FC", "Gamma FC", 3, 0 elif i % 4 == 1: h, a, hg, ag = "Delta FC", "Beta FC", 2, 0 elif i % 4 == 2: h, a, hg, ag = "Gamma FC", "Alpha FC", 0, 2 else: h, a, hg, ag = "Beta FC", "Delta FC", 0, 1 matches.append(FinishedMatch( str(i), "PL", dt, h, a, hg, ag, ids[h], ids[a], (h, h.replace(" FC", "")), (a, a.replace(" FC", "")), )) event = make_event() event["commence_time"] = (NOW + timedelta(hours=8)).isoformat() return matches, event def test_analyzer_market_prior_reduces_extreme_internal_probability(): from app.core.analyzer import analyze_events matches, event = _strong_favorite_fixture() picks, rejected = analyze_events( [event], matches, min_safe_score=70, limit=10, min_probability=.60, min_conservative_probability=.53, min_bookmakers=3, min_name_score=82, ) assert not rejected assert len(picks) == 1 pick = picks[0] internal = pick.raw_model_probability market = pick.market_probability posterior = pick.probability assert min(internal, market) <= posterior <= max(internal, market) assert pick.conservative_probability <= pick.probability assert pick.model_ev >= 0 def test_analyzer_rejects_single_bookmaker_for_safe_mode(): from app.core.analyzer import analyze_events from app.core.radar import build_radar matches, event = _strong_favorite_fixture() event["bookmakers"] = event["bookmakers"][:1] picks, rejected = analyze_events( [event], matches, min_safe_score=60, limit=10, min_probability=.55, min_conservative_probability=.50, min_bookmakers=3, min_name_score=80, ) assert not picks assert any("poucas casas" in row["reason"] for row in rejected) radar = build_radar(rejected) assert len(radar) == 1 assert radar[0]["approved"] is False assert radar[0]["selection"] assert any("poucas casas" in blocker for blocker in radar[0]["blockers"]) def test_analyzer_does_not_republish_open_event_from_previous_version(): from app.core.analyzer import analyze_events matches, event = _strong_favorite_fixture() picks, rejected = analyze_events( [event], matches, min_safe_score=70, limit=10, previous_picks=[{ "event_id": "evt1", "model_version": "2.1-precision", "side": "home", "market_probability": 0.65, }], min_probability=.60, min_conservative_probability=.53, min_bookmakers=3, min_name_score=82, ) assert picks == [] assert any("versão anterior" in row["reason"] for row in rejected) def test_stale_bookmakers_are_removed_from_consensus(): event = make_event() for bookmaker in event["bookmakers"]: bookmaker["last_update"] = (NOW - timedelta(hours=30)).isoformat() market = market_consensus(event, max_age_hours=12) assert market.bookmakers == 0 assert market.stale_bookmakers == 4 def test_football_data_falls_back_when_regular_time_is_null(): item = { "id": 100, "utcDate": NOW.isoformat(), "competition": {"code": "PL"}, "homeTeam": {"id": 1, "name": "Home FC"}, "awayTeam": {"id": 2, "name": "Away FC"}, "score": { "regularTime": {"home": None, "away": None}, "fullTime": {"home": 2, "away": 1}, }, } match = FootballDataProvider._parse_match(item) assert match is not None assert (match.home_goals, match.away_goals) == (2, 1) def test_performance_metrics_can_isolate_model_version(): history = [ {"result": "win", "probability": .70, "profit_units": .5, "model_version": MODEL_VERSION}, {"result": "loss", "probability": .70, "profit_units": -1, "model_version": "old"}, ] metrics = performance_metrics(history, MODEL_VERSION) assert metrics["settled"] == 1 assert metrics["wins"] == 1 assert metrics["legacy_or_other_version_excluded"] == 1 def test_append_history_updates_kickoff_for_rescheduled_same_event(): history = [] base = { "event_id": "resched", "kickoff": NOW.isoformat(), "competition": "Premier League", "competition_code": "PL", "home": "Alpha", "away": "Beta", "selection": "Alpha", "side": "home", "odd": 1.5, "probability": .7, "safe_score": 85, "model_version": MODEL_VERSION, } append_new_picks(history, [base]) moved = dict(base) moved["kickoff"] = (NOW + timedelta(days=2)).isoformat() moved["selection"] = "Beta" moved["side"] = "away" append_new_picks(history, [moved]) assert len(history) == 1 assert history[0]["kickoff"] == moved["kickoff"] assert history[0]["selection"] == "Alpha" assert history[0]["side"] == "home" def test_walk_forward_tuning_never_passes_future_matches(monkeypatch): import app.core.stats as stats_module matches = sample_matches("PL", 140) original = stats_module.predictive_models checked = {"calls": 0} def guarded(home_key, away_key, train, elo, competition=None, as_of=None, ensemble_weights=None): assert as_of is not None assert all(m.utc_date < as_of for m in train) checked["calls"] += 1 return original( home_key, away_key, train, elo, competition=competition, as_of=as_of, ensemble_weights=ensemble_weights, ) monkeypatch.setattr(stats_module, "predictive_models", guarded) result = stats_module.tune_ensemble_weights(matches, "PL", evaluation_matches=30) assert checked["calls"] > 0 assert abs(sum(result["weights"]) - 1.0) < 1e-9 def test_logging_suppresses_httpx_info_to_protect_query_keys(): import logging from app.logging_config import configure_logging configure_logging() assert logging.getLogger("httpx").level >= logging.WARNING assert logging.getLogger("httpcore").level >= logging.WARNING def test_walk_forward_reports_time_safe_brier_skill(): from app.core.stats import tune_ensemble_weights result = tune_ensemble_weights(sample_matches("PL", 180), "PL", evaluation_matches=40) assert result["samples"] >= 18 assert result["validation_samples"] >= 6 assert result["validation_samples"] < result["samples"] assert result["climatology_brier"] > 0 assert result["brier"] > 0 assert -2.0 < result["brier_skill"] < 1.0 def test_http_admin_surface_is_post_only_and_protected(): from fastapi.testclient import TestClient from app.main import app with TestClient(app) as client: assert client.get("/api/health").status_code == 200 assert client.get("/api/state").status_code == 200 assert client.get("/").status_code == 200 assert client.get("/api/cron/daily").status_code == 405 assert client.post("/api/cron/daily").status_code == 401