| # SPDX-License-Identifier: LicenseRef-PolyForm-Noncommercial-1.0.0 (see LICENSE)
|
| """Independent pandas/numpy reference for Session Levels MT5 (buffers 0-6), written from the spec, not from the MQL code.
|
|
|
| Assumes the server clock is New York close (server time = New York time + 7h), so the
|
| trading day is simply the server calendar day (17:00 New York = 00:00 server) and a New York wall-clock
|
| minute is (server seconds - 7h) mod a day. Bars are identified by their open time (epoch seconds, server time).
|
|
|
| b0 previous-day high b1 previous-day low b2 previous-week high b3 previous-week low
|
| b4 Asia high so far b5 Asia low so far b6 +1 first close > PDH of the day, -1 first close < PDL,
|
| +2 first close > Asia high after Asia ended, -2 first close < Asia low
|
| Days/weeks: week = Monday-based week of the trading day. The first day/week in the data has no "previous" (NaN).
|
| The last bar is the live bar: callers compare closed bars only (b6 of the last bar is NaN here as well)."""
|
| import numpy as np
|
| import pandas as pd
|
|
|
| NY_SHIFT = 7 * 3600
|
|
|
|
|
| def hm(s: str) -> int:
|
| h, m = s.split(":")
|
| return int(h) * 60 + int(m)
|
|
|
|
|
| def in_session(minute, s, e):
|
| if s == e:
|
| return np.zeros(len(minute), bool)
|
| return (minute >= s) & (minute < e) if s < e else (minute >= s) | (minute < e)
|
|
|
|
|
| def session_levels(t, o, h, l, c, asia_start="19:00", asia_end="04:00") -> np.ndarray:
|
| """Return an (n, 7) float array, NaN for empty."""
|
| t = np.asarray(t, dtype="int64")
|
| h, l, c = (np.asarray(x, float) for x in (h, l, c))
|
| n = len(t)
|
| out = np.full((n, 7), np.nan)
|
| day = t // 86400
|
| week = (day + 3) // 7 # 1970-01-01 was a Thursday; Monday-based week number
|
| df = pd.DataFrame({"day": day, "week": week, "h": h, "l": l, "c": c})
|
| # previous day / week extremes: aggregate each group, then look one group back
|
| for key, hi_col, lo_col in (("day", 0, 1), ("week", 2, 3)):
|
| agg = df.groupby(key).agg(hi=("h", "max"), lo=("l", "min"))
|
| prev = agg.shift(1)
|
| out[:, hi_col] = df[key].map(prev["hi"]).to_numpy()
|
| out[:, lo_col] = df[key].map(prev["lo"]).to_numpy()
|
| # Asia range so far, for the trading day the Asia session belongs to
|
| nyw = t - NY_SHIFT
|
| minute = (nyw % 86400) // 60
|
| s, e = hm(asia_start), hm(asia_end)
|
| ina = in_session(minute, s, e)
|
| owner = nyw // 86400 + ((minute >= s) & (s > e)).astype(int)
|
| a = pd.DataFrame({"owner": np.where(ina, owner, -1), "day": day, "h": h, "l": l})
|
| a_in = a[ina]
|
| run_hi = a_in.groupby("owner").h.cummax()
|
| run_lo = a_in.groupby("owner").l.cummin()
|
| first_day = a_in.groupby("owner").day.transform("first")
|
| ah = pd.Series(np.nan, index=a.index); al = ah.copy(); ad = ah.copy()
|
| ah[ina], al[ina], ad[ina] = run_hi, run_lo, first_day
|
| ah, al, ad = ah.ffill(), al.ffill(), ad.ffill() # the latest Asia instance carries forward
|
| valid = (ad.to_numpy() == day)
|
| out[:, 4] = np.where(valid, ah.to_numpy(), np.nan)
|
| out[:, 5] = np.where(valid, al.to_numpy(), np.nan)
|
| # signals on closed bars, each kind once per trading day
|
| pdh, pdl = out[:, 0], out[:, 1]
|
| after_asia = valid & ~ina
|
| for d in np.unique(day):
|
| idx = np.flatnonzero(day == d)
|
| idx = idx[idx < n - 1] # closed bars only
|
| done = set()
|
| for i in idx:
|
| kinds = []
|
| if not np.isnan(pdh[i]):
|
| if 1 not in done and c[i] > pdh[i]:
|
| kinds.append(1)
|
| if 2 not in done and c[i] < pdl[i]:
|
| kinds.append(2)
|
| if after_asia[i]:
|
| if 4 not in done and c[i] > out[i, 4]:
|
| kinds.append(4)
|
| if 8 not in done and c[i] < out[i, 5]:
|
| kinds.append(8)
|
| if kinds:
|
| done.update(kinds)
|
| k = kinds[0] # priority: PDH, PDL, Asia high, Asia low
|
| out[i, 6] = {1: 1, 2: -1, 4: 2, 8: -2}[k]
|
| return out
|
