piker/tests/test_trend_regimes.py

# piker: trading gear for hackers
# Copyright (C) 2018-present Tyler Goodlet (in stewardship of pikers)
# SPDX-License-Identifier: AGPL-3.0-or-later

# This program is free software: you can redistribute it and/or
# modify it under the terms of the GNU Affero General Public License
# as published by the Free Software Foundation, either version 3
# of the License, or (at your option) any later version.

# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.

# You should have received a copy of the GNU Affero General Public
# License along with this program. If not, see
# <https://www.gnu.org/licenses/>.

'''
Closed-bar evidence contracts for multiple-timeframe classifiers.

'''
from dataclasses import replace
import json

import numpy as np
import pytest

from piker.fsp._regime import (
    MultiTimeframeTrend,
    RegimeSnapshot,
    TimeframeRegime,
    TrendReversal,
)
from piker.fsp._trend import TrendCalculator, TrendUpdate


def test_timeframes_match_independent_closed_bar_calculators(
) -> None:
    '''
    Aggregating high/low extrema invents their intrabar order and can
    change TC reversals. Feed a zigzag scalar close stream through
    three periods and compare every prefix with separate calculators
    receiving only each complete bucket's final close. Exact trend,
    readiness, source-coordinate and detection-time agreement proves
    wall-clock aggregation does not change the TC recurrence.

    '''
    tracker: MultiTimeframeTrend = MultiTimeframeTrend(
        'synthetic.test', periods=(1, 2, 5), windows=(2, 2),
    )
    period: int
    engines: dict[int, TrendCalculator] = {
        period: TrendCalculator((2, 2)) for period in (1, 2, 5)
    }
    prices: np.ndarray = 100 + np.sin(np.arange(120) / 3) * 10
    row: int
    price: np.float64
    for row, price in enumerate(prices):
        snapshot: RegimeSnapshot = tracker.push(
            float(row), float(price), index=1000 + row,
        )
        frame: TimeframeRegime
        for frame in snapshot.timeframes:
            period: int = int(frame.period)
            if (row + 1) % period:
                continue
            expected: TrendUpdate = engines[period].push(
                float(row + 1), float(price), index=1000 + row,
            )
            assert frame.trends == expected.trends
            assert frame.ready == expected.ready
            assert frame.as_of == row + 1
            assert frame.recent_closes[-1].index == 1000 + row
            if len(expected.reversals):
                actual: tuple[TrendReversal, ...] = (
                    frame.reversals[-len(expected.reversals):]
                )
                got: TrendReversal
                event: np.void
                for got, event in zip(actual, expected.reversals):
                    assert got.time == event['time']
                    assert got.detected_time == (
                        event['detected_time']
                    )
                    assert got.index == event['index']
                    assert got.direction == event['direction']


def test_partial_startup_bucket_is_never_an_observation() -> None:
    '''
    Starting a feed mid-bucket must not give the first slow timeframe
    a shorter initial observation. Start a five-second stream at t=2
    and verify its [0, 5) bucket is explicitly discarded while the
    complete [5, 10) bucket yields its final close at t=10 only.

    '''
    tracker: MultiTimeframeTrend = MultiTimeframeTrend(
        'synthetic.test', periods=(5,),
    )
    second: int
    for second in range(2, 10):
        snapshot: RegimeSnapshot = tracker.push(second, second + 1)
        frame: TimeframeRegime = snapshot.timeframes[0]
        if second < 9:
            assert frame.close_count == 0
            assert frame.as_of is None
        else:
            assert frame.close_count == 1
            assert frame.as_of == 10
            assert frame.recent_closes[-1].price == 10
            assert frame.discarded_buckets == 1


def test_gap_resets_every_timeframe_and_rejects_partial_bucket(
) -> None:
    '''
    Carrying recurrence state across missing base bars fabricates
    continuity and can falsely report a ready regime. Warm two
    timeframes, omit t=9, and resume at t=10. The new source segment
    must clear all prior evidence, and the interrupted four-second
    bucket must not become a valid observation or restore readiness.

    '''
    tracker: MultiTimeframeTrend = MultiTimeframeTrend(
        'synthetic.test', periods=(1, 4), windows=(2,),
    )
    second: int
    for second in range(9):
        tracker.push(second, second)
    snapshot: RegimeSnapshot = tracker.push(10, 20)
    assert snapshot.segment == 1
    assert snapshot.timeframes[0].close_count == 1
    assert snapshot.timeframes[0].ready == (False,)
    assert snapshot.timeframes[1].close_count == 0
    assert not snapshot.timeframes[1].reversals
    snapshot = tracker.push(11, 21)
    assert snapshot.timeframes[1].close_count == 0
    for second in range(12, 16):
        snapshot = tracker.push(second, second)
    assert snapshot.timeframes[1].close_count == 1
    assert snapshot.timeframes[1].as_of == 16
    assert snapshot.timeframes[1].ready == (False,)


def test_subsecond_periods_preserve_epoch_alignment() -> None:
    '''
    Integer-second resampling silently truncates high-frequency
    periods, while float division can select the wrong epoch bucket.
    Four quarter-second base bars at a modern Unix epoch must form
    exactly one one-second close, with no early slow observation.

    '''
    tracker: MultiTimeframeTrend = MultiTimeframeTrend(
        'synthetic.test', base_period=.25, periods=(.25, 1),
    )
    offset: int
    for offset in range(4):
        snapshot: RegimeSnapshot = tracker.push(
            1_700_000_000 + offset / 4,
            100 + offset,
        )
        assert snapshot.timeframes[1].close_count == (offset == 3)
    assert snapshot.as_of == 1_700_000_001
    assert snapshot.timeframes[1].recent_closes[-1].price == 103


def test_decimal_periods_accept_float_roundoff() -> None:
    '''
    A calculated duration such as 0.1 * 3 differs from 0.3 by one
    float rounding step. Rejecting that representation prevents a
    valid integer-microsecond bucket configuration. Normalize only
    float-rounding noise and verify three tenth-second samples form
    one three-tenth-second observation at the expected endpoint.

    '''
    tracker: MultiTimeframeTrend = MultiTimeframeTrend(
        'test.market', base_period=.1, periods=(.1, .1 * 3),
    )
    tracker.push(0, 1)
    tracker.push(.1, 2)
    snapshot: RegimeSnapshot = tracker.push(.2, 3)
    assert snapshot.timeframes[1].period == .3
    assert snapshot.timeframes[1].as_of == .3
    assert snapshot.timeframes[1].close_count == 1


@pytest.mark.parametrize('kwargs', [
    {'base_period': 0},
    {'base_period': float('nan')},
    {'base_period': 0.0000001},
    {'base_period': 2, 'periods': (1,)},
    {'periods': (1, 1)},
    {'periods': ()},
    {'periods': (1.5,)},
    {'windows': (1,)},
    {'history_limit': 0},
    {'history_limit': 257},
    {'revision': -1},
])
def test_invalid_parameters_reject_ambiguous_sampling(
    kwargs: dict,
) -> None:
    '''
    A timeframe that cannot be built from whole base intervals has
    no complete-bucket interpretation. Each invalid configuration
    must fail before constructing a usable stream; bounded history
    and valid recurrence windows are checked at the same boundary.

    '''
    with pytest.raises(ValueError):
        MultiTimeframeTrend('synthetic.test', **kwargs)


@pytest.mark.parametrize('bar', [
    (0, 100, 1),
    (-1, 100, 1),
    (1.25, 100, 1),
    (1, float('nan'), 1),
    (1, float('inf'), 1),
    (float('inf'), 100, 1),
    (1, 100, 0),
    (1, 100, 2 ** 63),
    (1, 100, np.uint64(2 ** 63)),
    (1, 100, np.bool_(True)),
    (1e308, 100, 1),
])
def test_invalid_bar_leaves_all_timeframes_unchanged(
    bar: tuple[float, float, int|np.integer|np.bool_],
) -> None:
    '''
    Partial mutation before rejecting a bad bar causes different
    timeframes to consume different prefixes. Compare a tracker
    subjected to one invalid update with an untouched reference;
    accepting the same next bar must produce identical snapshots.

    '''
    tracker: MultiTimeframeTrend = MultiTimeframeTrend('test.market')
    reference: MultiTimeframeTrend = MultiTimeframeTrend(
        'test.market',
    )
    tracker.push(0, 100)
    reference.push(0, 100)
    with pytest.raises(ValueError):
        tracker.push(bar[0], bar[1], index=bar[2])
    assert tracker.push(1, 101) == reference.push(1, 101)


def test_revision_replay_rebuilds_causal_evidence() -> None:
    '''
    Revised history cannot be prepended to a warm TC recurrence.
    Replaying the identical complete stream after reset must retain
    every observation and reversal while changing its revision; the
    frozen snapshot from the earlier generation must stay unchanged.

    '''
    tracker: MultiTimeframeTrend = MultiTimeframeTrend(
        'test.market', periods=(1, 2), windows=(2, 2),
    )
    prices: list[int] = [1, 2, 3, 4, 3, 2, 1, 0] * 5
    second: int
    price: int
    for second, price in enumerate(prices):
        original: RegimeSnapshot = tracker.push(second, price)
    original_state: str = json.dumps(original.to_state())
    tracker.reset()
    for second, price in enumerate(prices):
        replayed: RegimeSnapshot = tracker.push(second, price)
    assert replayed == replace(original, revision=1)
    assert json.dumps(original.to_state()) == original_state
    frame: TimeframeRegime
    assert any(frame.reversals for frame in replayed.timeframes)
    for frame in replayed.timeframes:
        reversal: TrendReversal
        for reversal in frame.reversals:
            assert reversal.time <= reversal.detected_time
            assert reversal.detected_time <= frame.as_of
    with pytest.raises(ValueError):
        tracker.reset(revision=1)


def test_snapshot_history_is_bounded_and_finite_json() -> None:
    '''
    Sending unbounded market history makes inference cost grow with
    uptime. A long oscillating stream must retain at most the chosen
    close/reversal tail per timeframe while preserving total count.
    Strict JSON encoding also proves no NaN/Infinity enters state.
    Accepted NumPy integer windows must normalize to native integers
    so configuration metadata cannot break that same JSON contract.

    '''
    tracker: MultiTimeframeTrend = MultiTimeframeTrend(
        'test.market', periods=(1, 2), history_limit=3,
        windows=(np.int64(2), np.int64(2), np.int64(2)),
    )
    second: int
    for second in range(200):
        snapshot: RegimeSnapshot = tracker.push(
            second,
            float(second % 7),
        )
    frame: TimeframeRegime
    for frame in snapshot.timeframes:
        assert len(frame.recent_closes) == 3
        assert len(frame.reversals) <= 3
        assert frame.close_count == 200 / frame.period
    decoded: dict = json.loads(json.dumps(
        snapshot.to_state(), allow_nan=False,
    ))
    assert decoded['timestamp_convention'] == 'interval_end'
    assert decoded['algorithm'] == 'trend_calculus'
    assert decoded['windows'] == [2, 2, 2]