import sys import unittest from pathlib import Path import pandas as pd SCRIPTS_DIR = Path(__file__).resolve().parents[1] sys.path.insert(0, str(SCRIPTS_DIR)) import authentic_raw_signal_coverage as mod # noqa: E402 class DailyEligibilityTest(unittest.TestCase): def test_requires_all_twenty_nine_calendar_days(self): dates = pd.date_range("2025-12-03", "2025-12-31", freq="1D", tz="UTC") closes = pd.Series(range(1, 30), index=dates, dtype=float) item = mod.daily_eligibility( pair="BTC/USDT:USDT", daily_close=closes, listing_time=pd.Timestamp("2019-01-01", tz="UTC"), signal_eligible_time=pd.Timestamp("2026-01-01", tz="UTC"), ) self.assertTrue(item.eligible) self.assertEqual(item.start_endpoint, pd.Timestamp("2025-12-03", tz="UTC")) self.assertEqual(item.end_endpoint, pd.Timestamp("2025-12-31", tz="UTC")) def test_missing_middle_day_is_not_complete(self): dates = pd.date_range("2025-12-03", "2025-12-31", freq="1D", tz="UTC") closes = pd.Series(range(1, 30), index=dates, dtype=float).drop( pd.Timestamp("2025-12-15", tz="UTC") ) item = mod.daily_eligibility( pair="BTC/USDT:USDT", daily_close=closes, listing_time=pd.Timestamp("2019-01-01", tz="UTC"), signal_eligible_time=pd.Timestamp("2026-01-01", tz="UTC"), ) self.assertFalse(item.complete) self.assertFalse(item.eligible) def test_nonpositive_endpoint_is_ineligible(self): dates = pd.date_range("2025-12-03", "2025-12-31", freq="1D", tz="UTC") closes = pd.Series(range(1, 30), index=dates, dtype=float) closes.iloc[-1] = 0.0 item = mod.daily_eligibility( pair="BTC/USDT:USDT", daily_close=closes, listing_time=pd.Timestamp("2019-01-01", tz="UTC"), signal_eligible_time=pd.Timestamp("2026-01-01", tz="UTC"), ) self.assertFalse(item.positive_finite) self.assertFalse(item.eligible) def test_power_partial_listing_day_is_excluded(self): closes = pd.Series( range(1, 31), index=pd.date_range("2025-12-06", "2026-01-04", freq="1D", tz="UTC"), dtype=float, ) listing_time = pd.Timestamp("2025-12-06T09:00:00Z") includes_partial_day = mod.daily_eligibility( pair="POWER/USDT:USDT", daily_close=closes, listing_time=listing_time, signal_eligible_time=pd.Timestamp("2026-01-04T12:00:00Z"), ) first_full_window = mod.daily_eligibility( pair="POWER/USDT:USDT", daily_close=closes, listing_time=listing_time, signal_eligible_time=pd.Timestamp("2026-01-05T00:00:00Z"), ) self.assertFalse(includes_partial_day.window_after_listing) self.assertFalse(includes_partial_day.complete) self.assertFalse(includes_partial_day.eligible) self.assertTrue(first_full_window.window_after_listing) self.assertTrue(first_full_window.eligible) self.assertEqual( first_full_window.start_endpoint, pd.Timestamp("2025-12-07T00:00:00Z"), ) def test_edge_partial_listing_day_is_excluded(self): closes = pd.Series( range(1, 31), index=pd.date_range("2026-03-19", "2026-04-17", freq="1D", tz="UTC"), dtype=float, ) listing_time = pd.Timestamp("2026-03-19T14:00:00Z") includes_partial_day = mod.daily_eligibility( pair="EDGE/USDT:USDT", daily_close=closes, listing_time=listing_time, signal_eligible_time=pd.Timestamp("2026-04-17T12:00:00Z"), ) first_full_window = mod.daily_eligibility( pair="EDGE/USDT:USDT", daily_close=closes, listing_time=listing_time, signal_eligible_time=pd.Timestamp("2026-04-18T00:00:00Z"), ) self.assertFalse(includes_partial_day.window_after_listing) self.assertFalse(includes_partial_day.eligible) self.assertTrue(first_full_window.window_after_listing) self.assertTrue(first_full_window.eligible) self.assertEqual( first_full_window.start_endpoint, pd.Timestamp("2026-03-20T00:00:00Z"), ) class RankingTest(unittest.TestCase): @staticmethod def item(pair: str, score: float, eligible: bool = True): return mod.DailyEligibility( pair=pair, listing_eligible=eligible, window_after_listing=eligible, complete=eligible, positive_finite=eligible, score=score if eligible else None, start_endpoint=pd.Timestamp("2025-12-03", tz="UTC"), end_endpoint=pd.Timestamp("2025-12-31", tz="UTC"), ) def test_exact_ties_use_canonical_pair(self): items = { f"P{index:02d}/USDT:USDT": self.item(f"P{index:02d}/USDT:USDT", 1.0) for index in range(10) } universe, ranks, valid = mod.rank_snapshot(items) self.assertTrue(valid) self.assertEqual(universe, sorted(universe)) self.assertEqual(ranks[sorted(universe)[0]], 1) def test_snapshot_needs_ten_eligible_pairs(self): items = { f"P{index}/USDT:USDT": self.item(f"P{index}/USDT:USDT", float(index)) for index in range(9) } _, _, valid = mod.rank_snapshot(items) self.assertFalse(valid) def test_long_top4_and_short_bottom4(self): items = { f"P{index:02d}/USDT:USDT": self.item( f"P{index:02d}/USDT:USDT", float(12 - index) ) for index in range(12) } universe, ranks, valid = mod.rank_snapshot(items) long = mod.xsmom_decision( pair="P00/USDT:USDT", side="long", item=items["P00/USDT:USDT"], universe=universe, ranks=ranks, snapshot_valid=valid, ) short = mod.xsmom_decision( pair="P11/USDT:USDT", side="short", item=items["P11/USDT:USDT"], universe=universe, ranks=ranks, snapshot_valid=valid, ) self.assertEqual(long[:2], ("pass", "xsmom_top4")) self.assertEqual(short[:2], ("pass", "xsmom_bottom4")) class FundingCoverageTest(unittest.TestCase): def test_uses_only_finalized_events_at_or_before_signal(self): rates = pd.Series( [0.1, 0.2, 0.3, 0.4], index=pd.to_datetime( [ "2026-01-01T00:00:00Z", "2026-01-01T08:00:00Z", "2026-01-01T16:00:00Z", "2026-01-02T00:00:00Z", ] ), ) result = mod.funding_coverage( rates, pd.Timestamp("2026-01-01T20:00:00Z") ) self.assertTrue(result["funding_coverage_valid"]) self.assertEqual(result["funding_event_3_rate"], 0.3) def test_gap_larger_than_eight_hours_invalidates_coverage(self): rates = pd.Series( [0.1, 0.2, 0.3], index=pd.to_datetime( [ "2026-01-01T00:00:00Z", "2026-01-01T08:00:00Z", "2026-01-02T00:00:00Z", ] ), ) result = mod.funding_coverage( rates, pd.Timestamp("2026-01-02T01:00:00Z") ) self.assertFalse(result["funding_coverage_valid"]) class RawSignalTimeTest(unittest.TestCase): def test_signal_eligible_time_is_next_five_minute_boundary(self): class Strategy: def advise_indicators(self, frame, metadata): del metadata return frame def advise_entry(self, frame, metadata): del metadata frame["enter_short"] = 1 return frame rows = mod.extract_raw_signals( strategy=Strategy(), main_frames={ "BTC/USDT:USDT": pd.DataFrame( {"date": pd.to_datetime(["2025-12-31T23:55:00Z"])} ) }, window_start=pd.Timestamp("2026-01-01T00:00:00Z"), window_end=pd.Timestamp("2026-01-02T00:00:00Z"), ) self.assertEqual(rows[0]["signal_eligible_time"], "2026-01-01T00:00:00+00:00") if __name__ == "__main__": unittest.main()