forked from chiki2bum2/SniperGold_ML
Chiki Bum
0d5ca3c197
Implements the frozen P3_DATA_ENGINE_V1_SPEC (SHA 84bf0f217ffba51197112a6bbacbcc297058e04b5ca47f0459028fe33e0321e5). Components: engine/ producer (certify, chunkmap, parse, canonical, worker, dispatcher, aggregate, storage, journal, checkpoint, lock, evidence, manifest, run_complete, dataset_builder, cli) + engine/verify independent verifier (vparse, vaggregate, vinvariants, vcompare); headless CLI sniper-data; golden corpus G01-G17; unit/property/adversarial/mutation/legacy-diff/CLI suites. Qualification verdict: QUALIFIED (all 9 mandatory gates pass; independent verifier accepted). Spec, governance record, and legacy checkpoint untouched. G-14 NOT AUTHORIZED honored; no real-data processing, no pilot, no workload 46, no chunk 760 access.
182 lines
No EOL
6.8 KiB
Python
182 lines
No EOL
6.8 KiB
Python
"""Sequential deterministic bar aggregation (Layer 5, spec 11).
|
|
|
|
UTC clock boundaries; mid_u2 = bid_u + ask_u OHLC (PRICE_SCALE_BAR =
|
|
2*PRICE_SCALE); spread min/max/sum/avg (floor division); carry-aware; EOF
|
|
partial bars are flushed with is_final=False; no empty-bar inflation; bar_idx
|
|
is a global ordinal per timeframe; period_id strictly increases per TF which
|
|
implies (TF, period_id) uniqueness.
|
|
|
|
This module is the producer aggregator; engine/verify/vaggregate.py is the
|
|
independent counterpart and must agree on every row.
|
|
"""
|
|
|
|
from . import versions
|
|
|
|
|
|
def _new_carry():
|
|
return None
|
|
|
|
|
|
def start_carry(tick, period_id, period_ms):
|
|
ts, bid, ask, spread, vol, src_line, _rec = tick
|
|
mid = bid + ask
|
|
return {
|
|
"period_id": period_id,
|
|
"start_ms": period_id * period_ms,
|
|
"open_u2": mid,
|
|
"high_u2": mid,
|
|
"low_u2": mid,
|
|
"close_u2": mid,
|
|
"tick_count": 1,
|
|
"vol_sum": vol,
|
|
"spread_min_u": spread,
|
|
"spread_max_u": spread,
|
|
"spread_sum_u": spread,
|
|
"last_ts_ms": ts,
|
|
"last_src_line": src_line,
|
|
}
|
|
|
|
|
|
def update_carry(carry, tick):
|
|
ts, bid, ask, spread, vol, src_line, _rec = tick
|
|
mid = bid + ask
|
|
carry["high_u2"] = max(carry["high_u2"], mid)
|
|
carry["low_u2"] = min(carry["low_u2"], mid)
|
|
carry["close_u2"] = mid
|
|
carry["tick_count"] += 1
|
|
carry["vol_sum"] += vol
|
|
carry["spread_min_u"] = min(carry["spread_min_u"], spread)
|
|
carry["spread_max_u"] = max(carry["spread_max_u"], spread)
|
|
carry["spread_sum_u"] += spread
|
|
carry["last_ts_ms"] = ts
|
|
carry["last_src_line"] = src_line
|
|
return carry
|
|
|
|
|
|
def finalize_bar(carry, bar_idx, period_ms, is_final):
|
|
row = (
|
|
bar_idx,
|
|
carry["period_id"],
|
|
carry["start_ms"],
|
|
carry["start_ms"] + period_ms,
|
|
carry["open_u2"],
|
|
carry["high_u2"],
|
|
carry["low_u2"],
|
|
carry["close_u2"],
|
|
carry["tick_count"],
|
|
carry["vol_sum"],
|
|
carry["spread_min_u"],
|
|
carry["spread_max_u"],
|
|
carry["spread_sum_u"],
|
|
carry["spread_sum_u"] // carry["tick_count"], # floor; tick_count >= 1
|
|
is_final,
|
|
carry.get("first_src_line") if "first_src_line" in carry else None,
|
|
carry["last_src_line"],
|
|
)
|
|
return row
|
|
|
|
|
|
class Aggregator:
|
|
"""Carry-aware sequential aggregator over ordered canonical ticks."""
|
|
|
|
def __init__(self, cfg, carries=None, bar_counts=None, last_final_period=None):
|
|
self.cfg = cfg
|
|
tfs = list(cfg["timeframes"])
|
|
self.timeframes = tfs
|
|
self.periods = {tf: versions.TIMEFRAMES[tf] for tf in tfs}
|
|
self.carries = dict(carries or {})
|
|
self.bar_counts = dict(bar_counts or {tf: 0 for tf in tfs})
|
|
self.last_final_period = dict(last_final_period or {})
|
|
self._buckets = {tf: [] for tf in tfs}
|
|
self._resume_checked = False
|
|
|
|
# -- streaming ----------------------------------------------------------
|
|
def consume(self, tick):
|
|
"""Feed one canonical tick (ts,bid,ask,spread,vol,src_line,rec_ord).
|
|
|
|
On the first tick after construction with restored carries, spec 11.4
|
|
requires first input ts_ms >= every carried last_ts_ms; a violation is
|
|
a material carry mismatch (FAILED, F11).
|
|
"""
|
|
ts = tick[0]
|
|
if not self._resume_checked:
|
|
for tf in self.timeframes:
|
|
c = self.carries.get(tf)
|
|
if c is not None and ts < c.get("last_ts_ms", -1):
|
|
raise ValueError(
|
|
"carry continuity violation (%s): first ts %d < "
|
|
"carried last_ts %d" % (tf, ts, c.get("last_ts_ms")))
|
|
self._resume_checked = True
|
|
for tf in self.timeframes:
|
|
period_ms = self.periods[tf]
|
|
period_id = ts // period_ms
|
|
carry = self.carries.get(tf)
|
|
if carry is None:
|
|
self.carries[tf] = start_carry(tick, period_id, period_ms)
|
|
self.carries[tf]["first_src_line"] = tick[5]
|
|
continue
|
|
if period_id == carry["period_id"]:
|
|
update_carry(carry, tick)
|
|
continue
|
|
# period advanced -> finalize the old bar
|
|
if period_id < carry["period_id"]:
|
|
raise ValueError("aggregation order violation: period regressed")
|
|
last = self.last_final_period.get(tf)
|
|
if last is not None and carry["period_id"] <= last:
|
|
raise ValueError("(TF, period_id) duplicate finalization")
|
|
self.last_final_period[tf] = carry["period_id"]
|
|
row = finalize_bar(carry, self.bar_counts[tf], period_ms, True)
|
|
self.buckets_tf(tf).append(row)
|
|
self.bar_counts[tf] += 1
|
|
self.carries[tf] = start_carry(tick, period_id, period_ms)
|
|
self.carries[tf]["first_src_line"] = tick[5]
|
|
|
|
def buckets_tf(self, tf):
|
|
return self._buckets[tf]
|
|
|
|
def finish_workload(self, tf):
|
|
"""Return and clear the finalized-bar bucket for one timeframe."""
|
|
rows = self._buckets[tf]
|
|
self._buckets[tf] = []
|
|
return rows
|
|
|
|
def finish_eof(self, tf):
|
|
"""Flush the EOF partial bar (is_final=False) into the bucket, then
|
|
return and clear the bucket. Empty-bar inflation never happens: a
|
|
carry exists only when the timeframe saw at least one tick."""
|
|
carry = self.carries.get(tf)
|
|
if carry is not None:
|
|
last = self.last_final_period.get(tf)
|
|
if last is not None and carry["period_id"] <= last:
|
|
raise ValueError("(TF, period_id) duplicate finalization at EOF")
|
|
self.last_final_period[tf] = carry["period_id"]
|
|
row = finalize_bar(carry, self.bar_counts[tf], self.periods[tf], False)
|
|
self._buckets[tf].append(row)
|
|
self.bar_counts[tf] += 1
|
|
self.carries[tf] = None
|
|
rows = self._buckets[tf]
|
|
self._buckets[tf] = []
|
|
return rows
|
|
|
|
def has_partial(self, tf):
|
|
return self.carries.get(tf) is not None
|
|
|
|
def carr_state(self):
|
|
"""Serializable carry state for checkpointing."""
|
|
return {tf: (dict(self.carries.get(tf))
|
|
if self.carries.get(tf) else None)
|
|
for tf in self.timeframes}
|
|
|
|
def validate_resume_continuity(self):
|
|
"""Cross-check carries vs bar counts (spec 11.5 / 13.5).
|
|
|
|
bar_idx counts must be >= number of finalized bars; each carry's
|
|
period must be strictly greater than the last finalized period.
|
|
Raises ValueError (material F11) on mismatch."""
|
|
for tf in self.timeframes:
|
|
carry = self.carries.get(tf)
|
|
if carry is None:
|
|
continue
|
|
last = self.last_final_period.get(tf)
|
|
if last is not None and carry["period_id"] <= last:
|
|
raise ValueError("carry/%s continuity mismatch" % tf) |