signal-platform/tests/unit/test_residual_normalization.py

"""Economic invariants and calendar/coverage guards for normalization research."""

import json
import sqlite3
from datetime import date, timedelta
from pathlib import Path

import numpy as np
import pytest

from app.services.momentum_service import compute_residual_12_1_momentum
from scripts import run_residual_normalization as study


def sessions(n=400):
    days, day = [], date(2019, 1, 2)
    while len(days) < n:
        if day.weekday() < 5:
            days.append(day)
        day += timedelta(days=1)
    return days


def series(beta=1.3, scale=1.0):
    days = sessions()
    rng = np.random.default_rng(59)
    market = rng.normal(0.0003, 0.01, len(days) - 1)
    specific = rng.normal(0.0008, 0.005, len(days) - 1)
    spy = [100.0, *list(100 * np.cumprod(1 + market))]
    stock = [100.0, *list(100 * np.cumprod(1 + beta * market + scale * specific))]
    return days, stock, dict(zip(days, spy)), {d: i for i, d in enumerate(days)}


def test_numerator_is_production_and_denominator_uses_sample_variance():
    days, stock, spy, index = series()
    i = 300
    got = study.signal_at(days, stock, i, spy, index)
    assert got["paired_control"] == compute_residual_12_1_momentum(days[:i + 1], stock[:i + 1], spy)
    s = np.diff(np.array(stock[i - 252:i - 21 + 1])) / np.array(stock[i - 252:i - 21])
    mclose = np.array([spy[d] for d in days[i - 252:i - 21 + 1]])
    m = np.diff(mclose) / mclose[:-1]
    # Independent least-squares slope calculation, keeping alpha in the score.
    intercept, beta = np.linalg.lstsq(np.column_stack([np.ones(len(m)), m]), s, rcond=None)[0]
    assert intercept != 0
    expected_sd = (s - beta * m).std(ddof=1)
    assert got["residual_sd"] == pytest.approx(expected_sd)
    assert got["normalized"] == pytest.approx(got["paired_control"] / expected_sd)


def test_normalization_removes_specific_return_scale_but_keeps_direction():
    days, stock, spy, index = series(scale=1)
    _, larger, _, _ = series(scale=3)
    a = study.signal_at(days, stock, 350, spy, index)
    b = study.signal_at(days, larger, 350, spy, index)
    assert abs(a["paired_control"]) > 0.01
    assert b["paired_control"] == pytest.approx(3 * a["paired_control"])
    assert b["normalized"] == pytest.approx(a["normalized"])
    assert np.sign(a["normalized"]) == np.sign(a["paired_control"])


def test_market_beta_change_does_not_become_specific_strength():
    days, stock, spy, index = series(beta=0.5)
    _, high_beta, _, _ = series(beta=2.5)
    a = study.signal_at(days, stock, 350, spy, index)
    b = study.signal_at(days, high_beta, 350, spy, index)
    assert b["normalized"] == pytest.approx(a["normalized"])


def test_skipped_month_and_future_are_not_signal_inputs():
    days, stock, spy, index = series()
    i = 300
    baseline = study.signal_at(days, stock, i, spy, index)
    for k in range(i - 20, len(stock)):
        stock[k] *= 1.8
        spy[days[k]] *= 0.7
    assert study.signal_at(days, stock, i, spy, index) == baseline


def test_missing_benchmark_pair_stays_only_in_unrestricted_control():
    days, stock, spy, index = series()
    del spy[days[200]]
    index = {d: i for i, d in enumerate(sorted(spy))}
    got = study.signal_at(days, stock, 350, spy, index)
    assert got["unrestricted_control"] is not None
    assert got["paired_control"] is None
    assert got["normalized"] is None


def test_missing_stock_session_cannot_be_called_daily_aligned():
    days, stock, spy, index = series()
    del days[200]
    del stock[200]
    got = study.signal_at(days, stock, 349, spy, index)
    assert got["unrestricted_control"] is not None
    assert got["reason"] == "unaligned_formation"


def test_perfect_market_tracker_is_not_an_infinite_score():
    days, _, spy, index = series()
    got = study.signal_at(days, [spy[d] for d in days], 350, spy, index)
    assert got["normalized"] is None
    assert got["reason"] == "zero_residual_variance"


def test_nonoverlap_uses_sessions_and_permits_only_a_shared_endpoint():
    # Arbitrary indices stand in for a calendar with holidays/year boundaries.
    candidates = [100, 104, 109, 114, 119, 123, 128, 133, 138, 143, 148, 153, 158, 163]
    assert study.nonoverlapping_indices(candidates) == [100, 133, 163]


def test_weekly_endpoint_not_moved_earlier_to_obtain_forward_data():
    days = sessions(100)
    got = study.weekly_indices(days)
    last_per_week = {}
    for i, day in enumerate(days):
        last_per_week[day.isocalendar()[:2]] = i
    assert all(last_per_week[days[i].isocalendar()[:2]] == i for i in got)
    assert all(i + study.HORIZON < len(days) for i in got)


def test_identical_columns_have_zero_paired_bootstrap_difference():
    x = np.linspace(-0.1, 0.2, 25)
    matrix = np.column_stack([x, x, x])
    got = study.bootstrap_means(matrix)
    assert got["paired_delta_ci95"] == [0.0, 0.0]
    assert got == study.bootstrap_means(matrix)


def test_screen_uses_unrounded_floor_and_requires_both_eras():
    def summaries(value, pre=0.01, post=0.04):
        return {era: {"windows": 20, "arms": {"normalized": {"mean_ic": mean}}}
                for era, mean in (("full", value), ("pre_2021", pre), ("post_2021", post))}
    assert not study.screen_pass(summaries(0.0299999))["passed"]
    assert study.screen_pass(summaries(0.03))["passed"]
    assert not study.screen_pass(summaries(0.04, pre=-0.001))["passed"]


def test_screen_keeps_paired_stocks_and_forward_endpoints_identical(monkeypatch):
    days = sessions()
    benchmark = {d: 100 + i for i, d in enumerate(days)}
    prices = {}
    for k in range(25):
        closes = [100 + k + i * (k + 1) / 100 for i in range(len(days))]
        prices[f"S{k:02}"] = ([d.toordinal() for d in days], [], [], [], closes, [])

    def fake_signal(dates, closes, i, spy, index):
        value = closes[0]
        # One name remains in the unrestricted control but not the paired set.
        excluded = value == 100
        return {"unrestricted_control": value, "paired_control": None if excluded else value,
                "normalized": None if excluded else value / 2,
                "reason": "unaligned_formation" if excluded else "eligible"}

    monkeypatch.setattr(study, "signal_at", fake_signal)
    report = study.run_screen(prices, benchmark)
    assert report["windows"]
    index = {d.isoformat(): i for i, d in enumerate(days)}
    previous_end = None
    for window in report["windows"]:
        assert index[window["forward_end"]] - index[window["asof"]] == 30
        if previous_end:
            assert window["asof"] >= previous_end
        previous_end = window["forward_end"]
        assert window["arms"]["unrestricted_control"]["n"] == 25
        assert window["arms"]["paired_control"]["n"] == 24
        assert window["arms"]["normalized"]["n"] == 24
        assert window["arms"]["normalized"]["ic"] == window["arms"]["paired_control"]["ic"]


def write_snapshot(path, closes):
    """Minimal snapshot: one ticker, SPY, and a matching completion manifest."""
    days = sessions(len(closes))
    con = sqlite3.connect(path)
    con.executescript(
        "CREATE TABLE tickers (id INTEGER, symbol TEXT);"
        "CREATE TABLE research_rank_only (symbol TEXT);"
        "CREATE TABLE ohlcv_records (ticker_id INTEGER, date TEXT, open REAL, high REAL,"
        " low REAL, close REAL, volume REAL);"
        "CREATE TABLE benchmark_prices (symbol TEXT, date TEXT, close REAL);"
        "INSERT INTO tickers VALUES (1, 'AAA');"
    )
    con.executemany(
        "INSERT INTO ohlcv_records VALUES (1, ?, ?, ?, ?, ?, NULL)",
        [(d.isoformat(), c, c, c, c) for d, c in zip(days, closes)],
    )
    con.executemany(
        "INSERT INTO benchmark_prices VALUES ('SPY', ?, 100.0)", [(d.isoformat(),) for d in days]
    )
    con.commit()
    con.close()
    Path(str(path) + ".manifest.json").write_text(json.dumps({
        "complete": True, "ticker_count": 1, "ohlcv_row_count": len(closes), "rank_only_count": 0,
    }))


def test_missing_price_stops_the_run_instead_of_reading_as_zero(tmp_path):
    path = tmp_path / "snap.sqlite"
    write_snapshot(path, [100.0, None, 102.0])
    with pytest.raises(ValueError, match="Missing OHLC price: AAA"):
        study.read_snapshot(path)


def test_snapshot_read_keeps_missing_volume_and_releases_the_file(tmp_path):
    path = tmp_path / "snap.sqlite"
    write_snapshot(path, [100.0, 101.0, 102.0])
    prices, benchmark, _ = study.read_snapshot(path)
    assert prices["AAA"][4] == [100.0, 101.0, 102.0]
    assert prices["AAA"][5] == [0.0, 0.0, 0.0]
    assert len(benchmark) == 3
    path.unlink()  # Windows refuses while a connection still holds the file.


def test_paired_coverage_mismatch_raises_rather_than_asserts(monkeypatch):
    days = sessions()
    benchmark = {d: 100 + i for i, d in enumerate(days)}
    prices = {
        f"S{k:02}": ([d.toordinal() for d in days], [], [], [],
                     [100 + k + i * (k + 1) / 100 for i in range(len(days))], [])
        for k in range(25)
    }

    def broken_pairing(dates, closes, i, spy, index):
        value = closes[0]
        return {"unrestricted_control": value, "paired_control": value,
                "normalized": None if value == 100 else value / 2, "reason": "eligible"}

    monkeypatch.setattr(study, "signal_at", broken_pairing)
    with pytest.raises(AssertionError, match="Paired arms scored different stocks"):
        study.run_screen(prices, benchmark)


def test_source_ids_are_git_blob_ids_whatever_the_line_endings(tmp_path):
    lf, crlf = tmp_path / "lf.txt", tmp_path / "crlf.txt"
    lf.write_bytes(b"hello\n")
    crlf.write_bytes(b"hello\r\n")
    # `printf 'hello\n' | git hash-object --stdin`
    assert study.git_blob_id(lf) == "ce013625030ba8dba906f756967f9e9ca394464a"
    assert study.git_blob_id(crlf) == study.git_blob_id(lf)