38 lines
1.4 KiB
Python
38 lines
1.4 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
|
|
from control_plane.trading_studio.indicators.batch01_oracle_manifest import (
|
|
batch01_oracle_request_manifest,
|
|
batch01_oracle_request_manifest_bytes,
|
|
)
|
|
from control_plane.trading_studio.indicators.historical_band_channel import (
|
|
OBSERVED_BAND_CHANNEL_PARAMS,
|
|
)
|
|
|
|
|
|
def test_batch01_oracle_manifest_has_exact_observed_requests_and_semantic_hashes():
|
|
manifest = batch01_oracle_request_manifest()
|
|
requests = manifest["requests"]
|
|
assert manifest["artifact"] == "HISTORICAL_FEATURE_ORACLE_V1_REQUEST"
|
|
assert manifest["engine_revision"] == "code-5056feb"
|
|
assert len(requests) == 97
|
|
assert {
|
|
(row["indicator_id"], row["period"], row["p1"])
|
|
for row in requests
|
|
} == OBSERVED_BAND_CHANNEL_PARAMS
|
|
for row in requests:
|
|
semantic = {
|
|
"engine_revision": manifest["engine_revision"],
|
|
"input_columns": manifest["input_columns"],
|
|
"output_dtype": manifest["output_dtype"],
|
|
"window_policy": manifest["window_policy"],
|
|
"indicator_id": row["indicator_id"],
|
|
"period": row["period"],
|
|
"p1": row["p1"],
|
|
}
|
|
assert row["semantic_fingerprint"] == hashlib.sha256(
|
|
json.dumps(semantic, sort_keys=True, separators=(",", ":")).encode()
|
|
).hexdigest()
|
|
assert json.loads(batch01_oracle_request_manifest_bytes()) == manifest
|