fin-tracker/backend/tests/sources/test_moex_sync.py

"""MOEX sync: which window each instrument is asked for.

The bug these cover: with one cursor for the whole source, a paper that entered the ledger
after the first run was only ever fetched from the cursor, so the stretch between the day it
was first held and that run stayed empty forever — and the valuation series had no price for
days the portfolio demonstrably held it.
"""

from __future__ import annotations

from datetime import date, timedelta

import httpx
import pytest
from sqlalchemy import func, select

from factories import make_account, make_event, make_instrument, make_price
from fintracker.analytics import today_local
from fintracker.config import Settings
from fintracker.db import get_sessionmaker
from fintracker.models import (
    AssetClass,
    Benchmark,
    BenchmarkKind,
    EventKind,
    Instrument,
    PriceCoverage,
    PriceDaily,
)
from fintracker.sources.moex.client import BoardInfo
from fintracker.sources.moex.sync import (
    OVERLAP_DAYS,
    Coverage,
    MoexSource,
    Target,
    choose_board,
    fetch_window,
    history_legs,
    history_start,
    needs_backfill,
)

ISS = "https://iss.moex.com/iss"
TODAY = today_local()
LISTING = date(2013, 3, 25)
"""When the mocked TQBR started: the day a backfill now reaches back to."""


def monday_back(days: int) -> date:
    """A weekday roughly `days` ago — the mocked ISS answers on trading days only."""
    d = TODAY - timedelta(days=days)
    return d - timedelta(days=d.isoweekday() - 1)


def settings() -> Settings:
    return Settings()


def target(since: date) -> Target:
    from fintracker.models import AssetClass

    return Target(
        instrument_id=1,
        secid="SBER",
        since=since,
        nominal=None,
        board="TQBR",
        exchange="shares",
        asset_class=AssetClass.share,
    )


BOARD_COLUMNS = [
    "secid",
    "boardid",
    "market",
    "engine",
    "is_primary",
    "history_from",
    "history_till",
]


def mock_iss(mock_http, boards: list[list] | None = None):
    """ISS with a history that answers exactly the window it was asked for."""
    mock_http.get(url__startswith=f"{ISS}/securities/").mock(
        return_value=httpx.Response(
            200,
            json={
                "boards": {
                    "columns": BOARD_COLUMNS,
                    "data": boards
                    or [["SBER", "TQBR", "shares", "stock", 1, "2013-03-25", TODAY.isoformat()]],
                }
            },
        )
    )
    mock_http.get(url__startswith=f"{ISS}/engines/").mock(
        return_value=httpx.Response(
            200, json={"marketdata": {"columns": ["LAST"], "data": [[275.65]]}}
        )
    )

    def _history(request: httpx.Request) -> httpx.Response:
        since = date.fromisoformat(request.url.params["from"])
        until = date.fromisoformat(request.url.params["till"])
        start = int(request.url.params.get("start", 0))
        days = [
            since + timedelta(days=i)
            for i in range((until - since).days + 1)
            if (since + timedelta(days=i)).isoweekday() < 6
        ][start : start + 100]
        return httpx.Response(
            200,
            json={
                "history": {
                    "columns": ["TRADEDATE", "CLOSE", "CURRENCYID"],
                    "data": [[d.isoformat(), 100.0, "SUR"] for d in days],
                }
            },
        )

    return mock_http.get(url__startswith=f"{ISS}/history").mock(side_effect=_history)


async def seed(*, first_held: date, priced: list[date], asked_from: date | None) -> int:
    """A ledger position, the prices already stored for it, and how deep we have asked."""
    account_id = await make_account(name="Брокерский")
    instrument_id = await make_instrument(ticker="SBER")
    await make_event(
        first_held,
        account_id=account_id,
        kind=EventKind.buy,
        instrument_id=instrument_id,
        quantity="10",
        price="100",
        amount="-1000",
    )
    for d in priced:
        await make_price(d, instrument_id=instrument_id, close="100")
    if asked_from is not None:
        async with get_sessionmaker()() as session:
            session.add(
                PriceCoverage(instrument_id=instrument_id, source="moex", history_from=asked_from)
            )
            await session.commit()
    return instrument_id


def legs(route) -> list[tuple[str, date, date]]:
    """One entry per (board, window) asked for, pagination collapsed."""
    seen = []
    for call in route.calls:
        params = call.request.url.params
        board = call.request.url.path.split("/boards/")[1].split("/")[0]
        leg = (board, date.fromisoformat(params["from"]), date.fromisoformat(params["till"]))
        if leg not in seen:
            seen.append(leg)
    return seen


def windows(route) -> list[tuple[date, date]]:
    """One entry per history request, pagination collapsed."""
    seen = []
    for call in route.calls:
        params = call.request.url.params
        window = (date.fromisoformat(params["from"]), date.fromisoformat(params["till"]))
        if window not in seen:
            seen.append(window)
    return seen


async def stored_days(instrument_id: int) -> tuple[int, date, date]:
    async with get_sessionmaker()() as session:
        row = (
            await session.execute(
                select(func.count(), func.min(PriceDaily.d), func.max(PriceDaily.d)).where(
                    PriceDaily.instrument_id == instrument_id
                )
            )
        ).one()
    return row[0], row[1], row[2]


@pytest.mark.parametrize(
    ("history_from", "expected_backfill"),
    [(None, True), (date(2026, 1, 1), True), (date(2025, 1, 1), False)],
)
def test_backfill_is_due_until_the_first_held_day_has_been_asked_for(
    history_from, expected_backfill
):
    coverage = Coverage(last=TODAY, history_from=history_from)
    assert needs_backfill(target(date(2025, 1, 1)), coverage) is expected_backfill


def test_a_settled_history_only_re_reads_the_tail():
    """ISS revises the settlement price after the close, so the last days are read again."""
    since = TODAY - timedelta(days=400)
    coverage = Coverage(last=TODAY - timedelta(days=1), history_from=since)

    assert fetch_window(target(since), coverage, TODAY) == (
        TODAY - timedelta(days=1 + OVERLAP_DAYS),
        TODAY,
    )


def test_a_paper_without_any_price_is_asked_from_the_day_it_was_held():
    since = TODAY - timedelta(days=30)
    assert fetch_window(target(since), Coverage(), TODAY) == (since, TODAY)


def test_history_starts_at_the_listing_when_that_is_before_the_first_purchase():
    whole = board("TQBR", primary=True, since="2013-03-25", till="2026-09-17")
    assert history_start(date(2025, 7, 8), whole) == date(2013, 3, 25)


def test_history_starts_at_the_first_purchase_when_the_board_began_later():
    """A board the paper moved to has no history before the move — the old board supplies it."""
    assert history_start(date(2025, 7, 8), MOVED_TO) == date(2025, 7, 8)


def test_history_starts_at_the_first_purchase_when_the_board_publishes_no_range():
    unknown = board("TQBR", primary=True, since=None, till=None)
    assert history_start(date(2025, 7, 8), unknown) == date(2025, 7, 8)


async def test_a_hole_in_the_past_is_pulled_back_to_the_listing(app, mock_http, run_sync):
    """The TBRU@ case: held since spring, priced only from the day the source first saw it.

    The sweep goes back past the first purchase to the listing, so the chart on the
    instrument card has the paper's own history rather than only the stretch it was held.
    """
    first_held = monday_back(60)
    recent = [TODAY - timedelta(days=n) for n in (3, 2, 1)]
    instrument_id = await seed(first_held=first_held, priced=recent, asked_from=recent[0])

    history = mock_iss(mock_http)
    result = await run_sync(MoexSource(), settings=settings())

    assert windows(history) == [(LISTING, TODAY)]
    count, low, high = await stored_days(instrument_id)
    assert low == LISTING
    assert high >= recent[-1]
    assert count > len(recent)
    assert result.counts["backfilled"] == 1
    assert result.warnings == []


async def test_a_complete_history_only_re_reads_the_overlap_window(app, mock_http, run_sync):
    first_held = monday_back(60)
    priced = [
        first_held + timedelta(days=n)
        for n in range(61)
        if (first_held + timedelta(days=n)).isoweekday() < 6
    ]
    await seed(first_held=first_held, priced=priced, asked_from=LISTING)

    history = mock_iss(mock_http)
    result = await run_sync(MoexSource(), settings=settings())

    assert windows(history) == [(priced[-1] - timedelta(days=OVERLAP_DAYS), TODAY)]
    assert result.counts["backfilled"] == 0


async def test_a_second_run_backfills_once_and_stores_no_duplicates(app, mock_http, run_sync):
    first_held = monday_back(60)
    instrument_id = await seed(
        first_held=first_held, priced=[TODAY - timedelta(days=1)], asked_from=None
    )

    history = mock_iss(mock_http)
    first = await run_sync(MoexSource(), settings=settings())
    after_first, _, _ = await stored_days(instrument_id)

    second = await run_sync(MoexSource(), settings=settings())
    after_second, _, _ = await stored_days(instrument_id)

    assert first.counts["backfilled"] == 1
    assert second.counts["backfilled"] == 0  # asking once is what the coverage row remembers
    assert windows(history)[-1][0] > first_held
    assert after_second == after_first  # (instrument_id, d) is the key: the sync upserts


async def test_a_long_break_inside_a_settled_history_is_reported(app, mock_http, run_sync):
    """A paper that stopped trading and one that failed to download look alike in the data —
    so the gap is surfaced as a warning instead of being silently re-fetched every run."""
    first_held = monday_back(120)
    priced = [first_held, first_held + timedelta(days=1), TODAY - timedelta(days=1)]
    await seed(first_held=first_held, priced=priced, asked_from=LISTING)

    mock_iss(mock_http)
    result = await run_sync(MoexSource(), settings=settings())

    assert result.warnings == [
        f"SBER: разрыв в истории {priced[1].isoformat()}..{priced[2].isoformat()}"
    ]


async def test_a_break_before_the_first_purchase_is_not_reported(app, mock_http, run_sync):
    """The older history exists for the chart; a halt years before the purchase is not ours."""
    first_held = monday_back(60)
    held = [
        first_held + timedelta(days=n)
        for n in range(61)
        if (first_held + timedelta(days=n)).isoweekday() < 6
    ]
    await seed(first_held=first_held, priced=[LISTING, *held], asked_from=LISTING)

    mock_iss(mock_http)
    result = await run_sync(MoexSource(), settings=settings())

    assert result.warnings == []


def board(name: str, *, primary: bool, since: str | None, till: str | None) -> BoardInfo:
    return BoardInfo(
        secid="TBRU",
        board=name,
        market="shares",
        engine="stock",
        is_primary=primary,
        currency="SUR",
        history_from=date.fromisoformat(since) if since else None,
        history_till=date.fromisoformat(till) if till else None,
    )


MOVED_FROM = board("TQTF", primary=False, since="2021-07-12", till="2026-06-19")
MOVED_TO = board("TQBR", primary=True, since="2026-06-22", till="2026-09-17")


def test_a_remembered_board_that_stopped_trading_is_dropped():
    """The June 2026 fund move: keeping TQTF freezes a fund on the day it left that board."""
    assert choose_board([MOVED_FROM, MOVED_TO], "TQTF").board == "TQBR"


def test_a_remembered_board_that_is_still_current_is_kept():
    """A paper listed on several live boards keeps the board the instrument was resolved to."""
    small = board("SMAL", primary=False, since="2011-11-21", till="2026-09-17")
    assert choose_board([small, MOVED_TO], "SMAL").board == "SMAL"


def test_history_before_a_move_is_asked_of_the_board_the_paper_left():
    since, until = date(2025, 7, 8), date(2026, 9, 18)

    assert history_legs([MOVED_FROM, MOVED_TO], MOVED_TO, since, until) == [
        (MOVED_FROM, since, date(2026, 6, 21)),
        (MOVED_TO, date(2026, 6, 22), until),
    ]


def test_a_board_that_covers_the_whole_window_is_asked_alone():
    """SBER also trades on SMAL; pricing it from two boards at once would mix the two."""
    since = date(2025, 1, 1)
    whole = board("TQBR", primary=True, since="2013-03-25", till="2026-09-17")

    assert history_legs([whole, MOVED_FROM], whole, since, date(2026, 9, 18)) == [
        (whole, since, date(2026, 9, 18))
    ]


async def test_a_backfill_spans_both_sides_of_a_board_move(app, mock_http, run_sync):
    first_held = monday_back(120)
    move = TODAY - timedelta(days=30)
    instrument_id = await seed(first_held=first_held, priced=[], asked_from=None)
    history = mock_iss(
        mock_http,
        boards=[
            [
                "SBER",
                "TQTF",
                "shares",
                "stock",
                0,
                "2021-07-12",
                (move - timedelta(days=3)).isoformat(),
            ],
            ["SBER", "TQBR", "shares", "stock", 1, move.isoformat(), TODAY.isoformat()],
        ],
    )

    await run_sync(MoexSource(), settings=settings())

    assert legs(history) == [
        ("TQTF", first_held, move - timedelta(days=1)),
        ("TQBR", move, TODAY),
    ]
    count, low, high = await stored_days(instrument_id)
    assert low == first_held  # the stretch on the old board is stored under the same instrument
    assert high >= TODAY - timedelta(days=2)  # ... and the new board carries it to today
    assert count > 60


async def add_benchmark(code: str, *, source: str = "moex", active: bool = True) -> int:
    async with get_sessionmaker()() as session:
        benchmark = Benchmark(
            code=code,
            name=f"Индекс {code}",
            kind=BenchmarkKind.total_return,
            source=source,
            currency="RUB",
            is_default=False,
            is_active=active,
        )
        session.add(benchmark)
        await session.commit()
        return benchmark.id


async def benchmark_instrument(benchmark_id: int) -> Instrument | None:
    async with get_sessionmaker()() as session:
        benchmark = await session.get(Benchmark, benchmark_id)
        assert benchmark is not None
        if benchmark.instrument_id is None:
            return None
        return await session.get(Instrument, benchmark.instrument_id)


async def test_an_active_benchmark_gets_an_instrument_and_its_whole_history(
    app, mock_http, run_sync
):
    """MCFTR is on RTSI, not SNDX with the other indices — the board comes from ISS."""
    benchmark_id = await add_benchmark("MCFTR")
    listed = date(2003, 2, 26)
    history = mock_iss(
        mock_http,
        boards=[["MCFTR", "RTSI", "index", "stock", 1, listed.isoformat(), TODAY.isoformat()]],
    )

    result = await run_sync(MoexSource(), settings=settings())

    instrument = await benchmark_instrument(benchmark_id)
    assert instrument is not None
    assert instrument.asset_class == AssetClass.market_index
    assert (instrument.ticker, instrument.board, instrument.exchange) == (
        "MCFTR",
        "RTSI",
        "index",
    )
    assert legs(history) == [("RTSI", listed, TODAY)]
    count, low, high = await stored_days(instrument.id)
    assert low == listed
    assert high >= TODAY - timedelta(days=3)
    assert count > 1000
    assert result.counts["last"] == 0  # an index has no live quote to keep
    assert result.warnings == []


async def test_a_second_run_does_not_duplicate_the_benchmark_instrument(app, mock_http, run_sync):
    benchmark_id = await add_benchmark("IMOEX")
    mock_iss(
        mock_http,
        boards=[["IMOEX", "SNDX", "index", "stock", 1, "1997-09-22", TODAY.isoformat()]],
    )

    await run_sync(MoexSource(), settings=settings())
    first = await benchmark_instrument(benchmark_id)
    second_run = await run_sync(MoexSource(), settings=settings())
    second = await benchmark_instrument(benchmark_id)

    assert first is not None and second is not None
    assert first.id == second.id
    assert second_run.counts["backfilled"] == 0


async def test_an_inactive_or_manual_benchmark_is_left_alone(app, mock_http, run_sync):
    inactive = await add_benchmark("RGBITR", active=False)
    manual = await add_benchmark("SPX", source="manual")

    result = await run_sync(MoexSource(), settings=settings())  # no ISS route: any call fails

    assert await benchmark_instrument(inactive) is None
    assert await benchmark_instrument(manual) is None
    assert result.counts == {"prices": 0}