forked from chiki2bum2/SniperGold_ML
110 lines
No EOL
4.4 KiB
Python
110 lines
No EOL
4.4 KiB
Python
"""Worker entry point (Layer 4, spec 10).
|
|
|
|
One chunk per worker process. The worker reads a byte range of the source
|
|
(read-only), parses it to canonical records, writes a staging parquet, fsyncs,
|
|
computes the content id and physical SHA-256, renames into the final ticks
|
|
directory and returns a stats payload. Workers never share memory; they
|
|
communicate with the parent only through files and their return value.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import time
|
|
|
|
from . import versions
|
|
from .canonical import (
|
|
tick_chunk_content_id, preservation_chunk_content_id, serialize_preservation,
|
|
)
|
|
from .config import validate_config
|
|
from .parse import parse_chunk
|
|
from .storage import ticks_chunk_path, ticks_staging_path, malformed_chunk_path, preservation_chunk_path, write_ticks_chunk
|
|
from .util import (
|
|
read_binary_range, sha256_file, sha256_bytes,
|
|
)
|
|
|
|
UTF8_BOM = b"\xef\xbb\xbf"
|
|
|
|
|
|
def _serialize_keys(records):
|
|
"""(ts_ms, bid_u, ask_u, vol) per canonical serialization rule (spec 8.5)."""
|
|
return [(r[0], r[1], r[2], r[4]) for r in records]
|
|
|
|
|
|
def _now_iso():
|
|
import datetime
|
|
return datetime.datetime.now(datetime.timezone.utc).isoformat().replace("+00:00", "Z")
|
|
|
|
|
|
def worker_main(config_path, source_path, out_root, source_id,
|
|
chunk_index, byte_start, byte_end, global_line_start,
|
|
expect_header, certify_time_ms):
|
|
"""Spawn-safe worker body. Returns a stats dict (picklable)."""
|
|
with open(config_path, "r", encoding="utf-8") as fh:
|
|
cfg = json.load(fh)
|
|
validate_config(cfg)
|
|
|
|
started = time.time()
|
|
data = read_binary_range(source_path, byte_start, byte_end)
|
|
if chunk_index == 0 and data.startswith(UTF8_BOM):
|
|
data = data[len(UTF8_BOM):]
|
|
|
|
records, malformed, counters, flags = parse_chunk(
|
|
data, chunk_index=chunk_index, byte_start=byte_start,
|
|
global_line_start=global_line_start, cfg=cfg,
|
|
expect_header=expect_header, certify_time_ms=certify_time_ms)
|
|
|
|
# Persist malformed sidecar (jsonl) before the canonical file rename so a
|
|
# crash mid-write leaves only an orphan .tmp (spec 10.4).
|
|
os.makedirs(os.path.dirname(ticks_staging_path(out_root, chunk_index)),
|
|
exist_ok=True)
|
|
if malformed:
|
|
mp = malformed_chunk_path(out_root, chunk_index)
|
|
os.makedirs(os.path.dirname(mp), exist_ok=True)
|
|
with open(mp, "wb") as fh:
|
|
for rec in malformed:
|
|
fh.write((json.dumps(rec, sort_keys=True,
|
|
separators=(",", ":")) + "\n").encode("utf-8"))
|
|
fh.flush()
|
|
os.fsync(fh.fileno())
|
|
|
|
# Source-preservation layer (spec 8.6): the G_TICKSTORY_MT5 ``last``
|
|
# column is persisted per canonical row before the canonical rename, so a
|
|
# crash mid-write leaves only an orphan .tmp. Dotted runs write nothing.
|
|
preservation_content_id = None
|
|
preservation_sha256 = None
|
|
if cfg["source_grammar"] == versions.GRAMMAR_TICKSTORY_MT5 and records:
|
|
last_u_list = flags.get("last_u") or []
|
|
if len(last_u_list) == len(records):
|
|
pp = preservation_chunk_path(out_root, chunk_index)
|
|
os.makedirs(os.path.dirname(pp), exist_ok=True)
|
|
payload = b"".join(
|
|
serialize_preservation(r[5], lu)
|
|
for r, lu in zip(records, last_u_list))
|
|
with open(pp, "wb") as fh:
|
|
fh.write(payload)
|
|
fh.flush()
|
|
os.fsync(fh.fileno())
|
|
preservation_content_id = preservation_chunk_content_id(
|
|
[(r[5], lu) for r, lu in zip(records, last_u_list)])
|
|
preservation_sha256 = sha256_file(pp)
|
|
staging = ticks_staging_path(out_root, chunk_index)
|
|
write_ticks_chunk(staging, records, cfg, source_id, chunk_index)
|
|
content_id = tick_chunk_content_id(_serialize_keys(records))
|
|
physical_sha = sha256_file(staging)
|
|
os.replace(staging, ticks_chunk_path(out_root, chunk_index))
|
|
|
|
stats = {
|
|
"chunk_index": chunk_index,
|
|
"rows_parsed": flags["rows_parsed"],
|
|
"is_header": flags["has_header"],
|
|
"canonical_rows": len(records),
|
|
"malformed_count": len(malformed),
|
|
"preservation_content_id": preservation_content_id,
|
|
"preservation_sha256": preservation_sha256,
|
|
"malformed": counters,
|
|
"content_id": content_id,
|
|
"file_sha256": physical_sha,
|
|
"elapsed_sec": time.time() - started,
|
|
"ts_utc": _now_iso(),
|
|
}
|
|
return stats |