hdc_validate_v3_test.py

import os
import pickle
import logging
import numpy as np
import csv

# ─── CONFIG ──────────────────────────────────────────────────────────────
TIMESCALE = "1d"
DATA_SPLIT = 0.9
DATA_DIR = f"datasets/hdc_pairs_single_shot/{TIMESCALE}"
MEMORY_FILE = os.path.join(DATA_DIR, "hdc_memory.pkl")
X_FILE = "X_pairs_shuffled.npy"
Y_FILE = "Y_pairs_shuffled.npy"
O_FILE = "O_pairs_shuffled.npy"

SAMPLE_SIZE = 100
VOLUME_K = 0.2
FUSION_OPTIONS = ["weighted", "mean"]
TOP_K_RANGE = range(1, 9)  # TOP_K from 1 to 8
AGGREGATION_METHOD = "SUM"  # "AVG" or "SUM"


TARGET_COLS = ["target_open", "target_high", "target_low", "target_close", "target_volume"]
ORIG_COLS = ["original_close", "original_volume", "original_quote_asset_volume",
             "original_number_of_trades", "original_taker_buy_base_asset_volume",
             "original_taker_buy_quote_asset_volume"]

OCLOSE, OVOLUME, OQUOTE, ONTRADES, OTBBAV, OTBQAV = range(6)

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")


# ─── HDC Helpers ─────────────────────────────────────────────────────────
def quantize_sequence(seq, edges):
    bins = np.empty_like(seq, dtype=np.int32)
    for f in range(seq.shape[1]):
        bins[:, f] = np.digitize(seq[:, f], edges[:, f])
    return bins


def bundle(vecs):
    return np.where(np.sum(vecs, axis=0) >= 0, 1, -1).astype(np.int8)


def bind(a, b):
    return (a * b).astype(np.int8)

def predict_sequence(mem, seq, top_k=5, fusion="weighted"):
    edges, feat_arr, time_arr = mem['edges'], mem['feature_tags_arr'], mem['time_tags_arr']
    keys, values, knorms = mem['keys'], mem['values'], mem['key_norms']

    bins = quantize_sequence(seq, edges)
    rows = []
    for t, bin_row in enumerate(bins):
        feat_vecs = feat_arr[np.arange(feat_arr.shape[0]), bin_row]
        rows.append(bind(time_arr[t], bundle(feat_vecs)))
    query = bundle(np.stack(rows, axis=0))

    sims = np.dot(keys, query) / (knorms * np.linalg.norm(query))
    top_id = np.argpartition(-sims, top_k)[:top_k]
    top_vectors = values[top_id]

    if fusion == "weighted":
        w = sims[top_id] / np.sum(sims[top_id])
        fused = (top_vectors * w[:, None]).sum(axis=0)
    elif fusion == "mean":
        fused = top_vectors.mean(axis=0)
    else:
        raise ValueError(f"Unsupported fusion type: {fusion}")

    return fused


# ─── Decoding ────────────────────────────────────────────────────────────
def decode_target_row(prev_close, prev_vals, row):
    out = {}
    out["open"] = prev_close * np.exp(row[0])
    out["high"] = prev_close * np.exp(row[1])
    out["low"] = prev_close * np.exp(row[2])
    out["close"] = prev_close * np.exp(row[3])
    vol_keys = ["volume"]
    for i, k in enumerate(vol_keys, start=4):
        out[k] = prev_vals[k] * np.exp(np.arctanh(row[i]) / VOLUME_K)
    return out


# ─── MAIN ────────────────────────────────────────────────────────────────
def run_experiments():
    mem = pickle.load(open(MEMORY_FILE, "rb"))
    X = np.load(os.path.join(DATA_DIR, X_FILE))
    Y = np.load(os.path.join(DATA_DIR, Y_FILE))
    O = np.load(os.path.join(DATA_DIR, O_FILE))

    N = X.shape[0]
    split = int(N * DATA_SPLIT)
    X_te, Y_te, O_te = X[split:], Y[split:], O[split:]
    logging.info(f"Loaded test‑set: {X_te.shape[0]} samples")

    rng = np.random.default_rng(seed=42)
    idxs = rng.choice(X_te.shape[0], size=SAMPLE_SIZE, replace=False)

    results = []

    for fusion in FUSION_OPTIONS:
        for top_k in TOP_K_RANGE:
            total_mse = {key: 0.0 for key in ["open", "high", "low", "close", "volume"]}
            for idx in idxs:
                feats = X_te[idx]
                target = Y_te[idx]
                orig = O_te[idx]

                synth = predict_sequence(mem, feats, top_k=top_k, fusion=fusion)

                prev_close = float(orig[OCLOSE])
                prev_vals = {"volume": float(orig[OVOLUME])}

                real_decoded = decode_target_row(prev_close, prev_vals.copy(), target)
                synth_decoded = decode_target_row(prev_close, prev_vals.copy(), synth)

                for key in total_mse:
                    real_val = real_decoded[key]
                    synth_val = synth_decoded[key]
                    total_mse[key] += (real_val - synth_val) ** 2

            if AGGREGATION_METHOD == "AVG":
                for key in total_mse:
                    total_mse[key] /= SAMPLE_SIZE

            results.append([
                fusion, top_k,
                total_mse["open"],
                total_mse["high"],
                total_mse["low"],
                total_mse["close"],
                total_mse["volume"],
            ])

    output_path = f"validation_reports/hdc_fusion_topk_report_{AGGREGATION_METHOD}.csv"
    with open(output_path, "w", newline="") as f:
        writer = csv.writer(f)
        writer.writerow(["fusion_type", "top_k", "open_mse", "high_mse", "low_mse", "close_mse", "volume_mse"])
        writer.writerows(results)

    return output_path

run_experiments()