SniperGold_ML/ml/p3/p3_s251_external_ingest/fixtures.py

# -*- coding: utf-8 -*-
"""P3-S25.1 FIXTURE BUILDER — deterministic synthetic Tickstory-style CSVs.

Coverage for the validation matrix: small counts, hour/day/session, DST
transition, weekend/gap, duplicate timestamps, malformed rows, M15/M30
boundaries. All deterministic; all CSV rows are byte-stable.
"""
import datetime as dt
import os

import s251_config as CFG

SCR = CFG.SCRATCH


def _row(date, hhmmss, bid, ask=None, last=None, vol=0):
    ask = ask if ask is not None else round(bid + 0.2, 3)
    last = last if last is not None else bid
    return "%s,%s,%.3f,%.3f,%.3f,%d\n" % (date, hhmmss, bid, ask, last, vol)


def _epoch(y, m, d, hh, mi, ss):
    return int(dt.datetime(y, m, d, hh, mi, ss, tzinfo=dt.timezone.utc).timestamp())


def ticks_100():
    """100 ticks over a few M15 buckets incl. boundaries + duplicate times."""
    rows = []
    base = _epoch(2017, 1, 3, 0, 0, 0)
    t = base
    p = 1200.0
    for i in range(100):
        y = dt.datetime.fromtimestamp(t, tz=dt.timezone.utc)
        date = "%04d%02d%02d" % (y.year, y.month, y.day)
        tm = "%02d:%02d:%02d" % (y.hour, y.minute, y.second)
        rows.append(_row(date, tm, p))
        p += 0.5 if i % 2 else -0.3
        t += 60 if i else 0  # duplicate first timestamp with second tick
    # ensure a 15-min boundary tick and an exact-boundary tick
    b15 = base + 900
    y = dt.datetime.fromtimestamp(b15, tz=dt.timezone.utc)
    rows.append(_row("%04d%02d%02d" % (y.year, y.month, y.day),
                     "%02d:%02d:%02d" % (y.hour, y.minute, y.second), 1250.0))
    return rows


def ticks_day():
    """One full trading day of ticks (dense, ~10k) with one session gap."""
    y0 = 2018, 4, 11
    base = _epoch(*y0, 0, 0, 0)
    rows = []
    t = base
    p = 1300.0
    gap_from = _epoch(*y0, 12, 0, 0)
    gap_to = _epoch(*y0, 13, 30, 0)
    i = 0
    while t < base + 86400:
        if gap_from <= t < gap_to:
            t = gap_to
            continue
        y = dt.datetime.fromtimestamp(t, tz=dt.timezone.utc)
        rows.append(_row("%04d%02d%02d" % (y.year, y.month, y.day),
                         "%02d:%02d:%02d" % (y.hour, y.minute, y.second), p))
        p += 0.3
        t += 600
        i += 1
    return rows


def ticks_dst():
    """Tick sequence crossing a DST 'spring forward' (2021-03-28 in EU)."""
    # EU DST 2021: 2021-03-28 01:00 UTC -> 03:00 CEST (01:00 skipped in local,
    # but our timestamps are UTC wall-clock, so we just ensure no shift).
    base = _epoch(2021, 3, 28, 0, 0, 0)
    rows = []
    for off in (-3600, -1800, 0, 1800, 3600, 5400, 7200, 9000):
        t = base + off
        y = dt.datetime.fromtimestamp(t, tz=dt.timezone.utc)
        rows.append(_row("%04d%02d%02d" % (y.year, y.month, y.day),
                         "%02d:%02d:%02d" % (y.hour, y.minute, y.second),
                         1400.0 + off))
    return rows


def ticks_weekend():
    """Friday close -> Mon open (weekend gap)."""
    fr = _epoch(2019, 11, 1, 21, 0, 0)   # Fri evening
    mo = _epoch(2019, 11, 4, 1, 0, 0)    # Mon early
    rows = []
    for t in (fr, fr + 600, mo, mo + 600, mo + 1800):
        y = dt.datetime.fromtimestamp(t, tz=dt.timezone.utc)
        rows.append(_row("%04d%02d%02d" % (y.year, y.month, y.day),
                         "%02d:%02d:%02d" % (y.hour, y.minute, y.second),
                         1500.0))
    return rows


def ticks_dupes():
    """Multiple ticks with identical timestamps."""
    base = _epoch(2019, 6, 3, 0, 0, 0)
    rows = []
    for i in range(6):
        rows.append(_row("20190603", "00:00:00", 1600.0 + i))
    for i in range(3):
        rows.append(_row("20190603", "00:15:00", 1610.0 + i))
    return rows


def ticks_malformed():
    rows = [
        "20190603,00:00:00,1600.0,1600.2,1600.0,1\n",
        "20190603,00:00:05,1601.0,1601.2,1601.0,1\n",          # valid
        "20190603,25:61:99,1602.0,1602.2,1602.0,1\n",          # bad time
        "20190603,00:00:10,1603.0,1603.2,notanum,1\n",         # non-numeric
        "20190613,00:00:20\r\n",                                # wrong cols
        "20190603,00:00:30,1609.0,1608.9,1609.0,1\n",           # bid>ask invalid
        "20190603,00:00:40,1610.0,1610.2,1610.0,1\n",           # valid
    ]
    return rows


def write_fixture(name, rows):
    path = os.path.join(CFG.SCRATCH, name)
    with open(path, "w", encoding="ascii", newline="\n") as f:
        for r in rows:
            f.write(r if r.endswith("\n") else r + "\n")
    return path