signal-platform/tests/unit/test_earnings_research.py

from datetime import date, timedelta

from scripts.backfill_earnings_events import _dedupe_bulk_rows, _windows
from scripts.import_dolthub_earnings import _align_symbol
from scripts.run_earnings_research import (
    _analyse_2a_trades,
    _build_sue_series,
    _mechanical_sue_grade,
)


def test_dolthub_alignment_is_monotonic_across_close_calendar_events() -> None:
    events = [
        {"announce_date": date(2020, 3, 17), "announce_time": "bmo"},
        {"announce_date": date(2020, 4, 30), "announce_time": "bmo"},
    ]
    periods = [
        {"period_end_date": date(2019, 12, 31)},
        {"period_end_date": date(2020, 3, 31)},
    ]
    matches, unmatched_events, unmatched_periods = _align_symbol(
        events, periods, max_lag_days=90, max_lead_days=14
    )
    assert matches == [(0, 0), (1, 1)]
    assert unmatched_events == []
    assert unmatched_periods == []


def test_dolthub_alignment_allows_fiscal_period_label_after_announcement() -> None:
    events = [
        {"announce_date": date(2023, 2, 28), "announce_time": "bmo"},
        {"announce_date": date(2023, 5, 23), "announce_time": "bmo"},
    ]
    periods = [
        {"period_end_date": date(2023, 2, 28)},
        {"period_end_date": date(2023, 5, 31)},
    ]
    matches, _, _ = _align_symbol(
        events, periods, max_lag_days=90, max_lead_days=14
    )
    assert matches == [(0, 0), (1, 1)]


def test_bulk_windows_cover_range_without_overlap() -> None:
    result = _windows(date(2020, 1, 1), date(2020, 1, 10), 4)
    assert result == [
        (date(2020, 1, 1), date(2020, 1, 4)),
        (date(2020, 1, 5), date(2020, 1, 8)),
        (date(2020, 1, 9), date(2020, 1, 10)),
    ]


def test_bulk_dedupe_prefers_more_complete_and_counts_restatement() -> None:
    rows = [
        {
            "symbol": "AAPL",
            "announce_date": "2024-01-01",
            "announce_time": None,
            "eps_estimate": 1.0,
            "eps_actual": 1.1,
            "revenue_estimate": None,
            "revenue_actual": None,
        },
        {
            "symbol": "AAPL",
            "announce_date": "2024-01-01",
            "announce_time": "amc",
            "eps_estimate": 1.0,
            "eps_actual": 1.2,
            "revenue_estimate": 10.0,
            "revenue_actual": 11.0,
        },
    ]
    deduped, duplicates, restated = _dedupe_bulk_rows(rows)
    assert duplicates == 1
    assert restated == 1
    assert deduped == [rows[1]]


def test_2a_uses_net_r_strict_hold_and_next_session_stop() -> None:
    calendar = [
        date(2024, 1, 2),
        date(2024, 1, 3),
        date(2024, 1, 4),
        date(2024, 1, 5),
        date(2024, 1, 8),
        date(2024, 1, 9),
    ]
    events = [
        {
            "symbol": "AAPL",
            "announce_date": date(2024, 1, 5),
        }
    ]
    trades = [
        {
            "symbol": "AAPL",
            "entry_date": "2024-01-03",
            "exit_date": "2024-01-08",
            "entry": 100.0,
            "initial_stop": 90.0,
            "fill": 90.0,
            "r": -1.0,
            "reason": "stop",
        },
        {
            "symbol": "MSFT",
            "entry_date": "2024-01-02",
            "exit_date": "2024-01-09",
            "entry": 100.0,
            "initial_stop": 90.0,
            "fill": 110.0,
            "r": 1.0,
            "reason": "time",
        },
    ]
    result = _analyse_2a_trades(
        trades, events, calendar, cost_per_side=0.001
    )
    assert result["q1_loss_concentration"]["losses_count"] == 1
    assert result["q1_loss_concentration"]["losses_with_announcement_count"] == 1
    assert (
        result["q2_entries_within_3_trading_days_before_announcement"][
            "pre_earnings"
        ]["count"]
        == 1
    )
    assert (
        result["q3_stop_exits_within_1_trading_day_after_announcement"][
            "stops_after_earnings"
        ]["count"]
        == 1
    )
    assert result["q1_loss_concentration"]["loss_definition"] == (
        "realized_net_R <= -1.0"
    )


def test_sue_needs_four_prior_surprises_and_starts_next_trading_day() -> None:
    dates = [date(2024, 1, 1) + timedelta(days=index) for index in range(100)]
    columns = (
        [value.toordinal() for value in dates],
        [100.0] * len(dates),
        [101.0] * len(dates),
        [99.0] * len(dates),
        [100.0] * len(dates),
        [1_000_000] * len(dates),
    )
    event_dates = [date(2024, 1, 2) + timedelta(days=10 * index) for index in range(5)]
    surprises = [0.1, -0.2, 0.3, -0.1, 0.4]
    events = {
        "AAPL": [
            {
                "announce_date": event_date,
                "eps_actual": 1.0 + surprise,
                "eps_estimate": 1.0,
            }
            for event_date, surprise in zip(event_dates, surprises)
        ]
    }
    series, counts = _build_sue_series(
        events, {"AAPL": columns}, use_price_fallback=False
    )
    first_live = event_dates[-1] + timedelta(days=1)
    assert first_live in series["AAPL"]
    assert event_dates[-1] not in series["AAPL"]
    assert counts["standard_scaled_events"] == 1
    assert counts["price_fallback_events"] == 0


def test_sue_uses_period_history_only_for_scaling() -> None:
    dates = [date(2020, 1, 1) + timedelta(days=index) for index in range(100)]
    columns = (
        [value.toordinal() for value in dates],
        [100.0] * len(dates),
        [101.0] * len(dates),
        [99.0] * len(dates),
        [100.0] * len(dates),
        [1_000_000] * len(dates),
    )
    event_date = date(2020, 2, 3)
    events = {
        "AAPL": [
            {
                "announce_date": event_date,
                "period_end_date": date(2019, 12, 31),
                "eps_actual": 1.4,
                "eps_estimate": 1.0,
            }
        ]
    }
    history = {
        "AAPL": [
            {
                "period_end_date": date(2018, 12, 31)
                + timedelta(days=90 * index),
                "eps_actual": 1.0 + surprise,
                "eps_estimate": 1.0,
            }
            for index, surprise in enumerate([0.1, -0.2, 0.3, -0.1])
        ]
    }
    series, counts = _build_sue_series(
        events,
        {"AAPL": columns},
        use_price_fallback=False,
        surprise_history_by_symbol=history,
    )
    assert event_date + timedelta(days=1) in series["AAPL"]
    assert event_date not in series["AAPL"]
    assert counts["events_scaled_from_period_history"] == 1


def test_sue_grade_requires_positive_both_eras() -> None:
    full = {"mean_ic": 0.03, "reliable": True}
    passed, stable = _mechanical_sue_grade(
        full, {"mean_ic": 0.01}, {"mean_ic": 0.02}
    )
    assert passed is True
    assert stable is True
    failed, stable = _mechanical_sue_grade(
        full, {"mean_ic": -0.01}, {"mean_ic": 0.02}
    )
    assert failed is False
    assert stable is False