313 lines
11 KiB
Python
313 lines
11 KiB
Python
# 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]
|