# piker: trading gear for hackers Copyright (C) 2018-present  Tyler
# Goodlet (in stewardship of pikers)

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

'''
Live sampler checks for timestamped startup trade backlogs.

'''
from types import SimpleNamespace

import numpy as np
import pytest
import trio

from piker.data._sampling import (
    sample_and_broadcast,
    trade_time_s,
)
from piker.data import _sampling


@pytest.mark.parametrize(
    ('raw', 'expected'),
    [
        (1_790_360_484, 1_790_360_484.),  # seconds
        (1_790_360_484_000, 1_790_360_484.),  # milliseconds
        (1_790_360_484_000_000, 1_790_360_484.),
        (1_790_360_484_000_000_000, 1_790_360_484.),
        (None, None),  # provider without a tick timestamp
    ],
)
def test_trade_time_s_accepts_common_epoch_units(
    raw: int|None,
    expected: float|None,
) -> None:
    assert trade_time_s({'time': raw}) == expected


@pytest.mark.parametrize(
    ('bar_time', 'current_time_ns'),
    [
        (
            1_790_360_484.,
            1_790_360_484_250_000_000,
        ),
        (
            1_790_360_485.,
            1_790_360_484_750_000_000,
        ),
    ],
)
def test_queued_trade_is_not_written_to_a_later_bar(
    bar_time: float,
    current_time_ns: int,
) -> None:
    '''
    A 13:21:19 IB trade queued until the 13:21:24 1s bar must not
    add its price or volume to either live SHM tail. A current tick
    in the same input channel still updates both bars.

    '''
    dtype = [
        ('time', 'f8'),
        ('open', 'f8'),
        ('high', 'f8'),
        ('low', 'f8'),
        ('close', 'f8'),
        ('volume', 'f8'),
    ]
    rt = np.array(
        [(bar_time, 30_918., 30_918.,
          30_918., 30_918., 19.)],
        dtype=dtype,
    )
    hist = np.array(
        [(1_790_360_460., 30_918., 30_918.,
          30_918., 30_918., 248.)],
        dtype=dtype,
    )
    rt_shm = SimpleNamespace(array=rt)
    hist_shm = SimpleNamespace(array=hist)
    bus = SimpleNamespace(get_subs=lambda key: set())

    async def replay() -> None:
        send, receive = trio.open_memory_channel(4)
        async with send:
            await send.send({
                'mnq.cme.20261218': {
                    'ticks': [{
                        'type': 'trade',
                        'time': 1_790_360_479_217_767_927,
                        'price': 30_923.,
                        'size': 12_013.,
                    }],
                    'volume': 2_222_721.,
                },
            })
            await send.send({
                'mnq.cme.20261218': {
                    'ticks': [{
                        'type': 'trade',
                        'time': current_time_ns,
                        'price': 30_918.25,
                        'size': 2.,
                    }],
                    'volume': 2_222_723.,
                },
            })
            await send.send({
                'mnq.cme.20261218': {
                    'ticks': [{
                        'type': 'trade',
                        'time': int(
                            (bar_time + 2.) * 1_000_000_000
                        ),
                        'price': 30_925.,
                        'size': 100.,
                    }],
                    'volume': 2_222_823.,
                },
            })
            await send.send({
                'mnq.cme.20261218': {
                    'ticks': [{
                        'type': 'trade',
                        'price': 30_918.5,
                        'size': 1.,
                    }],
                    'volume': 2_222_824.,
                },
            })
        async with receive:
            await sample_and_broadcast(
                bus,
                rt_shm,
                hist_shm,
                receive,
                'ib',
            )

    trio.run(replay)
    assert float(rt['high'][-1]) == 30_918.5
    assert float(rt['volume'][-1]) == 22.
    assert float(hist['high'][-1]) == 30_918.5
    assert float(hist['volume'][-1]) == 251.


def test_next_second_trade_waits_for_sampler_step(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    '''
    A trade for the next 1s slot must not update the older SHM bar.

    The independent `Sampler.increment_ohlc_buffer()` can lag the
    tick stream. Waiting avoids a misdated price but delays chart
    updates, so a warning must fire before the slot advances. Hold
    the old slot until that warning, then advance it and verify the
    trade reaches both SHM tails without being dropped.

    '''
    dtype = [
        ('time', 'f8'),
        ('open', 'f8'),
        ('high', 'f8'),
        ('low', 'f8'),
        ('close', 'f8'),
        ('volume', 'f8'),
    ]
    rt = np.array(
        [(1_790_360_484., 30_918., 30_918.,
          30_918., 30_918., 0.)],
        dtype=dtype,
    )
    hist = rt.copy()
    rt_shm = SimpleNamespace(array=rt)
    hist_shm = SimpleNamespace(array=hist)
    bus = SimpleNamespace(get_subs=lambda key: set())
    warnings: list[str] = []
    wait_started = trio.Event()

    def record_warning(message: str) -> None:
        warnings.append(message)
        wait_started.set()

    monkeypatch.setattr(
        _sampling.log,
        'warning',
        record_warning,
    )

    async def replay() -> None:
        send, receive = trio.open_memory_channel(1)
        async with send:
            await send.send({
                'mnq.cme.20261218': {
                    'ticks': [{
                        'type': 'trade',
                        'time': 1_790_360_485_250_000_000,
                        'price': 30_919.,
                        'size': 2.,
                    }],
                    'volume': 2_222_723.,
                },
            })

        async def step() -> None:
            with trio.fail_after(2):
                await wait_started.wait()
            rt['time'][-1] = 1_790_360_485.

        async with (
            receive,
            trio.open_nursery() as nursery,
        ):
            nursery.start_soon(step)
            await sample_and_broadcast(
                bus,
                rt_shm,
                hist_shm,
                receive,
                'ib',
            )

    trio.run(replay)
    assert float(rt['high'][-1]) == 30_919.
    assert float(rt['volume'][-1]) == 2.
    assert float(hist['volume'][-1]) == 2.
    assert len(warnings) == 1
    assert 'DELAYING LIVE TRADE SHM UPDATE' in warnings[0]