prop-rules-monitor/tests/check_day_start.py

# SPDX-License-Identifier: LicenseRef-PolyForm-Noncommercial-1.0.0 (see LICENSE)
"""Line-by-line port of PrgDayStart / PrgBrokerOffset / PrgZoneOffset (MonEngine.mqh), checked against zoneinfo:
for every hour 2020-2026, on each broker clock and reset zone, the reset instant must be the last moment the zone's
wall clock showed reset_hour:00 (converted back to server time).   python3 tests/check_day_start.py"""
from datetime import datetime, timedelta, timezone
from zoneinfo import ZoneInfo

def sunday_from(y, m, d, h):
    t = datetime(y, m, d, tzinfo=timezone.utc)
    return t + timedelta(days=(6 - t.weekday()) % 7, hours=h)     # MQL day_of_week: Sunday=0; Python Sunday=6

def us_dst(u): return sunday_from(u.year, 3, 8, 7) <= u < sunday_from(u.year, 11, 1, 6)
def eu_dst(u): return sunday_from(u.year, 3, 25, 1) <= u < sunday_from(u.year, 10, 25, 1)

def zone_off(z, u):
    return {"Prague": 2 if eu_dst(u) else 1, "London": 1 if eu_dst(u) else 0,
            "NewYork": -4 if us_dst(u) else -5, "UTC": 0}[z] * 3600

def broker_off(clock, srv):        # srv is a naive "server clock" value stored as UTC-aware datetime
    if clock == "EET": return (3 if eu_dst(srv - timedelta(hours=2)) else 2) * 3600
    return (3 if us_dst(srv - timedelta(hours=2)) else 2) * 3600

def floor_to_hour(t, h):
    k = int(t.timestamp()) - h * 3600
    return t - timedelta(seconds=k % 86400)

def day_start(clock, zone, h, srv):
    utc = srv - timedelta(seconds=broker_off(clock, srv))
    loc = utc + timedelta(seconds=zone_off(zone, utc))
    rl = floor_to_hour(loc, h)
    ru = rl - timedelta(seconds=zone_off(zone, rl - timedelta(seconds=zone_off(zone, utc))))
    return ru + timedelta(seconds=broker_off(clock, ru + timedelta(hours=2)))

TZ = {"Prague": "Europe/Prague", "London": "Europe/London", "NewYork": "America/New_York", "UTC": "UTC"}
SRV = {"NY": "America/New_York", "EET": "Europe/Athens"}      # NY close = New York + 7h; EET = Athens

def truth(clock, zone, h, srv):
    # server wall clock -> UTC
    if clock == "NY":
        ny = srv.replace(tzinfo=None) - timedelta(hours=7)
        utc = ny.replace(tzinfo=ZoneInfo("America/New_York")).astimezone(timezone.utc)
    else:
        utc = srv.replace(tzinfo=None).replace(tzinfo=ZoneInfo("Europe/Athens")).astimezone(timezone.utc)
    loc = utc.astimezone(ZoneInfo(TZ[zone]))
    r = loc.replace(hour=h, minute=0, second=0, microsecond=0)
    if r > loc: r -= timedelta(days=1)
    r = r.replace(tzinfo=None).replace(tzinfo=ZoneInfo(TZ[zone])).astimezone(timezone.utc)
    # UTC -> server wall clock
    if clock == "NY":
        s = r.astimezone(ZoneInfo("America/New_York")).replace(tzinfo=None) + timedelta(hours=7)
    else:
        s = r.astimezone(ZoneInfo("Europe/Athens")).replace(tzinfo=None)
    return s.replace(tzinfo=timezone.utc)

bad = total = 0
for clock in ("NY", "EET"):
    for zone, h in (("Prague", 0), ("UTC", 0), ("NewYork", 17), ("London", 22), ("Prague", 23)):
        t = datetime(2020, 1, 1, tzinfo=timezone.utc)
        n_bad = 0
        while t < datetime(2027, 1, 1, tzinfo=timezone.utc):
            for minute in (0, 30):
                s = t + timedelta(minutes=minute)
                total += 1
                if day_start(clock, zone, h, s) != truth(clock, zone, h, s):
                    n_bad += 1
                    if n_bad <= 3:
                        print(f"  {clock} {zone} {h:02d}h srv {s:%Y-%m-%d %H:%M}: got {day_start(clock, zone, h, s):%m-%d %H:%M} want {truth(clock, zone, h, s):%m-%d %H:%M}")
            t += timedelta(hours=1)
        bad += n_bad
        print(f"{clock:3} {zone:8} {h:02d}:00  mismatches {n_bad}")
print(f"total {total} checks, {bad} mismatches")