# 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/>.

'''
Independent batch FHLS oracle and pinned original-Lua observations.

The oracle materializes complete windows and an expanded sequence of
FHLS summaries. It shares neither compiled helpers nor mutable window
state with the production calculator. See docs/trend_calculus.rst.

'''
from dataclasses import dataclass
from itertools import product
import json
from pathlib import Path
from typing import TypedDict

import numpy as np
import pytest

from piker.fsp._trend import (
    TrendCalculator,
    TrendHistory,
    TrendUpdate,
)


class _LuaCase(TypedDict):
    '''
    Describe the input and observed events of one pinned Lua run.

    '''
    window: int
    values: list[float]
    events: list[list[int|float]]


class _LuaFixture(TypedDict):
    '''
    Type the captured Lua cases consumed by the parity regression.

    '''
    cases: list[_LuaCase]


@dataclass(frozen=True)
class Point:
    '''
    Retain a pivot's source position and earliest availability row.

    '''
    row: int
    price: float
    available: int


@dataclass(frozen=True)
class Window:
    '''
    Summarize an entire immutable observation window using extrema.

    '''
    high: Point
    low: Point
    available: int

    @classmethod
    def summarize(
        cls: type['Window'],
        points: list[Point],
    ) -> 'Window':
        '''
        Python min/max retain the earliest equal extremum.

        '''
        return cls(
            high=max(points, key=lambda point: point.price),
            low=min(points, key=lambda point: point.price),
            available=points[-1].available,
        )

    def compare(self, previous: 'Window') -> int:
        '''
        Compare extrema ordinally, avoiding floating subtraction.

        '''
        if (
            (
                self.high.price >= previous.high.price
                and self.low.price >= previous.low.price
            )
            and
            (
                self.high.price > previous.high.price
                or self.low.price > previous.low.price
            )
        ):
            return 1
        if (
            (
                self.high.price <= previous.high.price
                and self.low.price <= previous.low.price
            )
            and
            (
                self.high.price < previous.high.price
                or self.low.price < previous.low.price
            )
        ):
            return -1
        return 0


def reference_order(
    points: list[Point],
    size: int,
    *,
    lua_seed: bool = False,
) -> tuple[list[tuple[int, int]], list[tuple[Point, int]]]:
    '''
    Expand complex windows, then derive directions and pivot events.

    Direction changes from unknown zero establish a baseline and do
    not manufacture a reversal. ``Point.available`` carries detection
    availability from the closing observation, including at recursive
    orders.

    '''
    start: int  # noqa: F842
    windows: list[Window] = [
        Window.summarize(points[start:start + size])
        for start in range(0, len(points) - size + 1, size)
    ]
    if lua_seed and windows:
        last: Point = points[size - 1]
        windows.insert(0, Window(last, last, last.available))
    if len(windows) < 2:
        return [], []

    expanded: list[Window] = [windows[0]]
    current: Window
    previous: Window
    for previous, current in zip(windows, windows[1:]):
        if current.compare(previous) == 0:
            second: Point = max(
                (previous.low, previous.high),
                key=lambda point: point.row,
            )
            first: Point = min(
                (current.low, current.high),
                key=lambda point: point.row,
            )
            # Lua's synthetic bridge breaks equal-value ties toward
            # its second endpoint for high and first for low.
            bridge_high: Point
            bridge_low: Point
            bridge_low, bridge_high = sorted(
                [second, first], key=lambda point: point.price,
            )
            expanded.append(Window(
                high=bridge_high,
                low=bridge_low,
                available=current.available,
            ))
        expanded.append(current)

    trend: int = 0
    directions: list[tuple[int, int]] = []
    events: list[tuple[Point, int]] = []
    previous: Window
    current: Window
    for previous, current in zip(expanded, expanded[1:]):
        direction: int = current.compare(previous) or trend
        if trend and direction != trend:
            pivot: Point = (
                previous.low if direction == 1 else previous.high
            )
            events.append((
                Point(pivot.row, pivot.price, current.available),
                direction,
            ))
        trend = direction
        directions.append((current.available, trend))
    return directions, events


def reference_hierarchy(
    values: np.ndarray,
    sizes: tuple[int, ...],
    *,
    lua_seed: bool = False,
) -> tuple[
    np.ndarray,
    np.ndarray,
    list[tuple[int, int, int, int]],
]:
    '''
    Repeatedly reduce whole point sequences, separately per order.

    The returned event tuples are pivot row, direction, order and
    detection row. Sorting by detection row and order reproduces
    causal publication while retaining stable same-order ordering.

    '''
    row: int
    price: np.floating|np.integer  # noqa: F842
    points: list[Point] = [
        Point(row, float(price), row)
        for row, price in enumerate(values)
    ]
    trends: np.ndarray = np.zeros(
        (len(values), len(sizes)), dtype=np.int8,
    )
    ready: np.ndarray = np.zeros_like(trends, dtype=np.bool_)
    all_events: list[tuple[int, int, int, int]] = []
    level: int
    size: int
    for level, size in enumerate(sizes):
        directions: list[tuple[int, int]]
        events: list[tuple[Point, int]]
        directions, events = reference_order(
            points, size, lua_seed=lua_seed,
        )
        if len(points) >= 2 * size:
            ready[points[2 * size - 1].available:, level] = True
        row: int
        direction: int
        for row, direction in directions:
            trends[row:, level] = direction
        point: Point
        direction: int
        for point, direction in events:
            all_events.append((
                point.row, direction, level + 1, point.available,
            ))
        point: Point
        direction: int
        points = [point for point, direction in events]
    all_events.sort(key=lambda event: (event[3], event[2]))
    # Public direction remains unresolved until two real windows,
    # including when the research seed establishes internal state.
    trends[~ready] = 0
    return trends, ready, all_events


@pytest.mark.parametrize('sizes', [(2, 2), (3, 2), (4, 3)])
@pytest.mark.parametrize('lua_seed', [False, True])
def test_exhaustive_ternary_histories(
    sizes: tuple[int, ...],
    lua_seed: bool,
) -> None:
    '''
    Enumerate ties, inside/outside bars and unresolved flat startup.

    Every eight-point sequence over three prices is compared with
    independently expanded ``reference_hierarchy()`` windows.
    Six-point sequences also exercise the optional Lua seed, without
    importing its CLI's output-suppression bugs. Readiness still
    requires two actual complete windows. Comparing every
    ``TrendHistory`` field catches incorrect first/second-extremum
    ordering and warmup decisions that agreement between two calls
    into the same production kernel cannot detect.

    '''
    length: int = 6 if lua_seed else 8
    sequence: tuple[int, ...]
    for sequence in product((-1, 0, 1), repeat=length):
        values: np.ndarray = np.array(sequence, dtype=np.float64)
        expected: np.ndarray
        readiness: np.ndarray
        events: list[tuple[int, int, int, int]]
        (
            expected,
            readiness,
            events,
        ) = reference_hierarchy(
            values, sizes, lua_seed=lua_seed,
        )
        actual: TrendHistory = TrendCalculator(
            sizes, lua_seed=lua_seed,
        ).extend(
            values, np.arange(len(values), dtype=np.float64),
        )
        assert np.array_equal(actual.trends, expected), sequence
        assert np.array_equal(actual.ready, readiness), sequence
        pulses: np.ndarray = np.zeros_like(expected)
        prices: np.ndarray = np.full(
            (len(values), len(sizes), 2), np.nan,
        )
        times: np.ndarray = np.full_like(prices, np.nan)
        pivot: int
        direction: int
        order: int
        detected: int
        for (
            pivot,
            direction,
            order,
            detected,
        ) in events:
            side: int = int(direction == 1)
            pulses[detected, order - 1] = direction
            prices[detected, order - 1, side] = values[pivot]
            times[detected, order - 1, side] = pivot
        assert np.array_equal(actual.pulses, pulses), sequence
        assert np.array_equal(
            actual.pivot_prices, prices, equal_nan=True,
        ), sequence
        assert np.array_equal(
            actual.pivot_times, times, equal_nan=True,
        ), sequence


@pytest.mark.parametrize('sizes', [(2,) * 5, (3, 2, 4), (5, 3, 2)])
@pytest.mark.parametrize('seed', range(8))
@pytest.mark.parametrize('lua_seed', [False, True])
def test_random_recursive_events(
    sizes: tuple[int, ...],
    seed: int,
    lua_seed: bool,
) -> None:
    '''
    Preserve recursive pivot causality across irregular coordinates.

    Higher-order pivots occur earlier than their availability, often
    sharing one detection row. Rounded random walks create tied and
    engulfing windows; ``reference_hierarchy()`` predicts every
    event independently of ``TrendCalculator.push()``. Gapped source
    indices and nonuniform timestamps prove that propagation retains
    source identity and detection time. Field equality and complete
    ordered-event equality detect lost or backdated reversals.

    '''
    rng: np.random.Generator = np.random.default_rng(seed)
    values: np.ndarray = np.cumsum(rng.integers(-3, 4, 512))
    times: np.ndarray = 1000 + np.cumsum(rng.uniform(.1, 2, 512))
    indices: np.ndarray = 800 + np.cumsum(rng.integers(1, 4, 512))
    expected: np.ndarray
    readiness: np.ndarray
    events: list[tuple[int, int, int, int]]
    (
        expected,
        readiness,
        events,
    ) = reference_hierarchy(
        values, sizes, lua_seed=lua_seed,
    )
    calculator: TrendCalculator = TrendCalculator(
        sizes, lua_seed=lua_seed,
    )
    actual_events: list[tuple[int, int, int, int]] = []
    row: int
    value: np.integer
    for row, value in enumerate(values):
        update: TrendUpdate = calculator.push(
            float(times[row]), float(value), index=int(indices[row]),
        )
        assert update.trends == tuple(expected[row])
        assert update.ready == tuple(readiness[row])
        event: np.void
        for event in update.reversals:
            pivot_row: int = int(np.searchsorted(
                indices, event['index'],
            ))
            assert event['price'] == values[pivot_row]
            assert event['time'] == times[pivot_row]
            assert event['detected_time'] == times[row]
            assert event['detected_index'] == indices[row]
            assert pivot_row <= row
            actual_events.append((
                pivot_row, int(event['direction']),
                int(event['order']), row,
            ))
    assert actual_events == events


def test_original_lua_fixture() -> None:
    '''
    Compare production events with actual, unmodified upstream Lua.

    The fixture uses increasing initial windows to establish a trend
    before reversals, avoiding Lua's documented flat-start crash and
    initial simple-reversal suppression. Every later pivot, sign and
    earliest detection prefix must agree, including tied extrema.
    Comparing complete ``TrendUpdate.reversals`` sequences against
    captured Lua output checks an implementation independent of the
    Python oracle. Fixture provenance records the exact program and
    input hashes; this case does not claim parity at Lua's startup
    failures.

    '''
    path: Path = (
        Path(__file__).parent
        / '_inputs'
        / 'trend_calculus_lua.json'
    )
    fixture: _LuaFixture = json.loads(path.read_text())
    case: _LuaCase
    for case in fixture['cases']:
        calculator: TrendCalculator = TrendCalculator(
            (case['window'],),
        )
        events: list[list[int|float]] = []
        row: int
        value: float
        for row, value in enumerate(case['values']):
            update: TrendUpdate = calculator.push(float(row), value)
            event: np.void
            for event in update.reversals:
                events.append([
                    int(event['index']), float(event['price']),
                    int(event['direction']),
                    int(event['detected_index']),
                ])
        assert events == case['events'], case['window']