forked from chiki2bum2/SniperGold_ML
127 lines
No EOL
4.4 KiB
Python
127 lines
No EOL
4.4 KiB
Python
# -*- 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 |