14 Commits
Author SHA1 Message Date
mathias 2c9bc1be31 fix(ci): drop Build & Import / Deploy via GitOps jobs (jepa-fx-risk#19)
CD / Lint / Test / Vet (push) Successful in 5s
This is a research repo -- issue#6/#7 already decided CI should be
check-only (`task check` on push/PR), no build, no image push, no
deploy, since training runs are manual/GPU-bound and never CI-triggered.
That never actually landed in cd.yml: it still carried the original
go-web template's build+deploy jobs verbatim, still referencing the
pre-rename `hostexecutor` name (`./cmd/hostexecutor` doesn't exist --
real dirs are cmd/eval, cmd/jepa-fx-risk) and a `k3s/apps/hostexecutor`
deployment that was never created in infra (confirmed: no manifest
exists under either name). Once the check job's own Taskfile bug was
fixed, build started actually running and failing on the wrong path.
Removed both jobs -- check is the only gate this repo needs.
2026-07-24 15:28:46 +02:00
mathias 308f71b566 fix(ci): Taskfile YAML syntax error + staticcheck tagged-switch (jepa-fx-risk#19)
CD / Build & Import (push) Failing after 10s
CD / Deploy via GitOps (push) Has been skipped
CD / Lint / Test / Vet (push) Successful in 7s
Taskfile.yml line 46 had an unquoted Go-template `{{.VAR}}` inside a YAML
flow sequence (`cmds: [...]`) -- the literal braces broke YAML parsing
outright ("did not find expected ',' or ']'"), so `task check` (and thus
CI's push-triggered check job) failed before running a single command.
A manual `workflow_dispatch` re-run passed because the autoresearch
workflow never calls `task check` at all -- unrelated path, not an
env/secret difference as first suspected. Quoted the string.

Also fixed a staticcheck QF1002 in internal/eval/var.go: a boolean
switch comparing the same variable (n1) in every case is a tagged
switch in disguise -- converted to `switch n1 { case 0: ... case n: ...
}`.
2026-07-24 15:26:01 +02:00
mathias 6fb77f5263 fix(ci): drop empty LOOP_MODEL env that blanked the model → LLM 400
CD / Lint / Test / Vet (push) Failing after 3s
CD / Deploy via GitOps (push) Has been skipped
CD / Build & Import (push) Has been skipped
`LOOP_MODEL: ${{ inputs.model }}` sets the env var to "" when no model input is
given. An env var set-but-empty is NOT unset, so it overrode loop.py's
berget/gemma4-31b default, and every agent call POSTed "model":"" → HTTP 400
Bad Request. The loop ran but optimized nothing (best == baseline, delta 0).

Real overrides already flow through the conditional `--model` arg, so this env
line was redundant as well as harmful. Remove it; loop.py's default applies.
2026-06-29 19:34:57 +00:00
mathias 8eb0358c01 fix(loop): put project root on PYTHONPATH for train.py subprocess
CD / Lint / Test / Vet (push) Failing after 2s
CD / Build & Import (push) Has been skipped
CD / Deploy via GitOps (push) Has been skipped
train.py is copied into runs/<rq-id>/ by autoresearch_start.py, so when loop.py
executes it, sys.path[0] is the run dir — which has no scripts/. train.py's
frozen VaR-eval block does `from scripts.var_breach import ...`, which then
fails with ModuleNotFoundError: No module named 'scripts' on every loop run
(CI and the documented manual launch alike).

Fix in the harness, not the frozen train.py/scripts boundary: prepend the
project root (loop.py's own dir, where scripts/ lives) to the subprocess
PYTHONPATH. Verified red→green locally: the import fails without it and
resolves with it (scripts/ is an implicit namespace package, no __init__.py).
2026-06-29 21:32:25 +02:00
mathias 2d412790bf chore(ci): commit 59KB toy eurusd_daily.parquet as CI data fixture
CD / Lint / Test / Vet (push) Failing after 3s
CD / Build & Import (push) Has been skipped
CD / Deploy via GitOps (push) Has been skipped
The Autoresearch Loop workflow (phase-a-toy) runs train.py, which reads
data/processed/eurusd_daily.parquet. data/ is gitignored (DVC/MinIO policy, #10),
but CI checks out a fresh tree with no data, so the baseline run died with
FileNotFoundError. Force-add this single 59KB derived fixture (public EUR/USD
daily realized vol, 2019-2023, 1560 rows) so the toy loop is hermetic and fast.

This is a scoped exception to #10 for a CI fixture only — large/production
datasets still stay out of git per that policy.
2026-06-29 21:20:41 +02:00
mathias 8539ec3a85 fix(ci): install torch from cu130 index in venv step
CD / Lint / Test / Vet (push) Failing after 2s
CD / Deploy via GitOps (push) Has been skipped
CD / Build & Import (push) Has been skipped
train.py imports torch but the venv step only installed requirements.txt,
which deliberately excludes torch (must come from the cu130 wheel index for
koala's Blackwell sm_120, per the requirements.txt header). Baseline run died
with ModuleNotFoundError: No module named 'torch'. Add the documented install.
2026-06-29 19:14:51 +00:00
mathias c9b22db3b8 fix(ci): pin upload-artifact to @v3 — v4 artifact protocol unsupported on Gitea
CD / Lint / Test / Vet (push) Failing after 2s
CD / Build & Import (push) Has been skipped
CD / Deploy via GitOps (push) Has been skipped
Gitea's act_runner does not implement the @actions/artifact v2.0.0+ backend, so
upload-artifact@v4 fails the job with GHESNotSupportedError even when the loop
succeeds. v3 is the newest tag the Gitea runner supports.
2026-06-29 18:43:12 +00:00
mathias 731673061e fix(deps): histdata>=1.3 is unsatisfiable, pin to >=1.1 (newest on PyPI)
CD / Lint / Test / Vet (push) Failing after 3s
CD / Build & Import (push) Has been skipped
CD / Deploy via GitOps (push) Has been skipped
The Autoresearch Loop workflow failed at "Set up Python venv" on every run:
  ERROR: Could not find a version that satisfies the requirement histdata>=1.3
          (from versions: 1.0, 1.1)
histdata never published 1.3 — 1.1 is the newest release. Lower the floor so
pip can resolve and the loop can actually start.
2026-06-29 18:41:07 +00:00
mathiasandClaude Sonnet 4.6 54b1bc216d feat(ci): Gitea Actions workflow for autoresearch loop (Phase A automation)
CD / Lint / Test / Vet (push) Failing after 3s
CD / Build & Import (push) Has been skipped
CD / Deploy via GitOps (push) Has been skipped
workflow_dispatch with inputs: fixture, rq_id, iters, model override.
- Scaffolds run dir from fixtures/<fixture>.json via autoresearch_start.py
- Runs loop.py with LITELLM_KEY/LITELLM_BASE/NTFY_URL secrets
- Uploads STATUS.md + metrics.json + program.md as artifacts (30-day retention)
- Cleans stale run dir at start; venv created/reused per runner workspace
- timeout-minutes: 90 to cover multi-iter runs

Phase B (k8s Job + GPU isolation) tracked in jepa-fx-risk#15.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-06-27 17:20:44 +02:00
mathiasandClaude Sonnet 4.6 fa6dcaae0a chore: fixture backlog for Phase A toy run + gitignore cleanup
CD / Lint / Test / Vet (push) Failing after 3s
CD / Build & Import (push) Has been skipped
CD / Deploy via GitOps (push) Has been skipped
- fixtures/phase-a-toy.json — scaffold source for `autoresearch start`
  Phase A live run: val_vol_r2 metric, daily EUR/USD, HEPA encoder toy
- .gitignore: add runs/ (ephemeral run dirs), metrics.json, embeddings.json,
  HEARTBEAT, STATUS.md, pyc/__pycache__/.pytest_cache, eval binary
- untrack metrics.json (was a stale committed sample, now gitignored)

Launch on koala:
  python scripts/autoresearch_start.py fixtures/phase-a-toy.json phase-a-toy
  op run -- env LITELLM_KEY="$LITELLM_KEY" \
    python loop.py --run-dir runs/phase-a-toy --iters 3

Closes the last gate on jepa-fx-risk#11 Phase A (code shipped v1.4.0;
live run pending).

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-06-27 17:03:55 +02:00
mathiasandClaude Sonnet 4.6 68bf8f15c5 feat(eval): VaR breach rate metric (#12) + HMM regime detector (#13) — rq-04 prep
CD / Lint / Test / Vet (push) Failing after 2s
CD / Build & Import (push) Has been skipped
CD / Deploy via GitOps (push) Has been skipped
#12 — VaR_breach_rate_99_oos_regime_cond metric:
- internal/eval/var.go: VaRBreachRate() + kupiecPOF() + LinearProbePredict() (stdlib math only)
- internal/eval/var_test.go: 8 golden tests (zero/all breach, perfect calibration, boundary)
- cmd/eval/main.go: -metric var flag (no-leakage probe → VaR → Kupiec P)
- scripts/var_breach.py: Python equivalent with METRIC_KEY constant (13 TDD tests)
- train.py LOCKED VaR EVAL BLOCK: writes VaR_breach_rate_99_oos_regime_cond + kupiec_p to metrics.json
- Fixed bug: train.py used bare 'os' before import; now uses module-level '_os' consistently

#13 — HMM regime detector + JEPA conditioning seam:
- scripts/prepare_regime.py: GaussianHMM (diag, 3-state) on realized_vol; states sorted by mean vol
  (0=calm, 1=stressed, 2=crisis); deterministic (random_state=42); outputs eurusd_regime.parquet
- tests/test_regime.py: 11 TDD tests (dtype, states, determinism, vol sort, daily fallback)
- train.py: JEPA_ENABLE_REGIME toggle + REGIME CONDITIONING SEAM (concat baseline, agent-editable)
- requirements.txt: hmmlearn>=0.3, scikit-learn>=1.4

78 Python + all Go tests green.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-06-27 10:35:10 +02:00
mathiasandClaude Sonnet 4.6 65a58fcca2 feat(loop): add --run-dir isolation, heartbeat, ntfy-on-crash; scaffold start command
CD / Lint / Test / Vet (push) Failing after 1s
CD / Build & Import (push) Has been skipped
CD / Deploy via GitOps (push) Has been skipped
Closes jepa-fx-risk#11 Phase A.

- scripts/autoresearch_start.py: scaffold runs/<rq-id>/ from Council backlog leaf;
  fail-closed on non-autoresearch-ready; strips candidate_metric; writes program.md
  + run.json (provenance) + train.py copy. 19 TDD tests.
- loop.py: --run-dir flag redirects STATUS.md / metrics.json / HEARTBEAT / train.py
  into the run dir; METRICS_OUT env var passed to train subprocess so it writes
  metrics.json to the run dir; heartbeat file written each iter phase; ntfy-on-crash
  via NTFY_URL env var (best-effort).
- train.py: METRICS_OUT env var overrides metrics.json path (default unchanged).

Launch: LITELLM_KEY=xxx python loop.py --run-dir runs/rq-04

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-06-27 10:19:17 +02:00
mathiasandClaude Sonnet 4.6 d4b67943fb feat(multipair): lock best config + train:multipair Taskfile target
CD / Lint / Test / Vet (push) Failing after 2s
CD / Build & Import (push) Has been skipped
CD / Deploy via GitOps (push) Has been skipped
Best: USE_MULTIPAIR=1 D_MODEL=256 WINDOW=120 → phase1_r2=0.4377, val_vol_r2=0.3695
WINDOW sweep (multipair D=256): W=60→0.4009, W=120→0.4377, W=240→0.4277
D_MODEL=128 undercapacity for 10ch (val_vol_r2=0.1013); D=256 restores probe quality.
Single-pair default knobs unchanged (D=128 still optimal there).

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-06-26 14:08:25 +02:00
mathiasandClaude Sonnet 4.6 bd8962f997 feat(multipair): G10 multi-pair pipeline + USE_MULTIPAIR knob (Option C)
- prepare_hourly.py: parameterize PAIR env var; OUT_DEFAULT per-pair; load_m1_from_zips(pair=)
- prepare_multipair.py: inner-join 5-pair hourly parquets on datetime → wide parquet
  cols: datetime, {pair}_ret, {pair}_rv × n_pairs; eurusd_rv = target
- fetch_multipair.py: download GBPUSD/USDJPY/USDCHF/AUDUSD M1 2008-2023 from histdata
- train.py: USE_MULTIPAIR knob (JEPA_USE_MULTIPAIR=1); build() reads multipair parquet
  with n_channels = n_pairs × 2; target = eurusd_rv
- Taskfile: data:fetch:multipair, data:prepare:pair, data:prepare:multipair, data:test updated
- 7 new tests in test_multipair.py; 34/35 pass (1 SKIP until multipair parquet built)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-06-26 14:00:57 +02:00
23 changed files with 1639 additions and 166 deletions
+75
View File
@@ -0,0 +1,75 @@
name: Autoresearch Loop
on:
workflow_dispatch:
inputs:
fixture:
description: 'Fixture name in fixtures/ (without .json)'
required: true
default: 'phase-a-toy'
rq_id:
description: 'Run ID — defaults to fixture name if blank'
required: false
default: ''
iters:
description: 'Max iterations'
required: false
default: '3'
model:
description: 'LiteLLM model override (leave blank for default berget/gemma4-31b)'
required: false
default: ''
jobs:
run:
name: Autoresearch — ${{ inputs.fixture }}
runs-on: self-hosted
timeout-minutes: 90
steps:
- uses: actions/checkout@v4
- name: Resolve run ID
id: vars
run: |
RQ_ID="${{ inputs.rq_id }}"
[ -z "$RQ_ID" ] && RQ_ID="${{ inputs.fixture }}"
echo "rq_id=$RQ_ID" >> "$GITHUB_OUTPUT"
- name: Clean stale run dir
run: rm -rf "runs/${{ steps.vars.outputs.rq_id }}"
- name: Set up Python venv
run: |
[ -d .venv ] || python3 -m venv .venv
# torch must come from the cu130 wheel index (koala Blackwell sm_120);
# requirements.txt deliberately excludes it. Install it first.
.venv/bin/pip install -q torch --index-url https://download.pytorch.org/whl/cu130
.venv/bin/pip install -q -r requirements.txt
- name: Scaffold run dir
run: |
.venv/bin/python scripts/autoresearch_start.py \
"fixtures/${{ inputs.fixture }}.json" \
"${{ steps.vars.outputs.rq_id }}"
- name: Run autoresearch loop
env:
LITELLM_KEY: ${{ secrets.LITELLM_KEY }}
LITELLM_BASE: ${{ secrets.LITELLM_BASE }}
NTFY_URL: ${{ secrets.NTFY_URL }}
run: |
ARGS="--run-dir runs/${{ steps.vars.outputs.rq_id }} --iters ${{ inputs.iters }}"
[ -n "${{ inputs.model }}" ] && ARGS="$ARGS --model ${{ inputs.model }}"
.venv/bin/python loop.py $ARGS
- name: Upload run artifacts
if: always()
uses: https://gitea.com/actions/upload-artifact@v3
with:
name: run-${{ steps.vars.outputs.rq_id }}-${{ github.run_number }}
path: |
runs/${{ steps.vars.outputs.rq_id }}/STATUS.md
runs/${{ steps.vars.outputs.rq_id }}/metrics.json
runs/${{ steps.vars.outputs.rq_id }}/program.md
retention-days: 30
-85
View File
@@ -7,9 +7,6 @@ on:
pull_request:
branches: [main]
env:
IMAGE: hostexecutor
jobs:
check:
name: Lint / Test / Vet
@@ -31,85 +28,3 @@ jobs:
- name: Run checks
run: task check
build:
name: Build & Import
needs: check
runs-on: self-hosted
if: github.event_name != 'pull_request'
outputs:
image-tag: ${{ steps.meta.outputs.sha-tag }}
steps:
- uses: actions/checkout@v4
- name: Derive image tags
id: meta
run: |
SHA=$(git rev-parse --short HEAD)
echo "sha-tag=${SHA}" >> "$GITHUB_OUTPUT"
- name: Build and push to local registry
run: |
REGISTRY="localhost:5000"
REF="${REGISTRY}/${{ env.IMAGE }}:${{ steps.meta.outputs.sha-tag }}"
buildah build \
--label "org.opencontainers.image.revision=${{ github.sha }}" \
-t ${REF} \
-t ${REGISTRY}/${{ env.IMAGE }}:latest \
.
buildah push --tls-verify=false ${REF}
buildah push --tls-verify=false ${REGISTRY}/${{ env.IMAGE }}:latest
echo "✓ Image pushed to ${REF}"
deploy:
name: Deploy via GitOps
needs: build
runs-on: self-hosted
if: github.ref == 'refs/heads/main' && github.event_name == 'push'
steps:
- name: Update image tag in infra repo
env:
IMAGE_TAG: ${{ needs.build.outputs.image-tag }}
DEPLOY_KEY: ${{ secrets.INFRA_DEPLOY_KEY }}
run: |
set -euo pipefail
mkdir -p ~/.ssh
echo "$DEPLOY_KEY" > ~/.ssh/id_infra
chmod 600 ~/.ssh/id_infra
ssh-keyscan -p 30022 10.0.1.20 >> ~/.ssh/known_hosts 2>/dev/null
export GIT_SSH_COMMAND="ssh -i ~/.ssh/id_infra -o IdentitiesOnly=yes"
rm -rf /tmp/infra
git clone -b main ssh://git@10.0.1.20:30022/mathias/infra.git /tmp/infra
cd /tmp/infra
DEPLOYMENT="k3s/apps/hostexecutor/deployment.yaml"
sed -i "s|image: localhost:5000/hostexecutor:.*|image: localhost:5000/hostexecutor:${IMAGE_TAG}|" "$DEPLOYMENT"
grep -q "localhost:5000/hostexecutor:${IMAGE_TAG}" "$DEPLOYMENT" \
|| { echo "✗ image tag patch failed"; exit 1; }
if git diff --quiet "$DEPLOYMENT"; then
echo " image tag unchanged — skipping push"
else
git -c user.name="hostexecutor CI" \
-c user.email="ci@hostexecutor.local" \
commit -m "chore(deploy): hostexecutor → ${IMAGE_TAG}" "$DEPLOYMENT"
git push origin main
echo "✓ pushed to infra repo"
fi
shred -u ~/.ssh/id_infra
- name: Trigger Flux reconcile
run: |
kubectl -n flux-system annotate gitrepository flux-system \
reconcile.fluxcd.io/requestedAt="$(date +%s)" --overwrite
kubectl -n flux-system annotate kustomization apps \
reconcile.fluxcd.io/requestedAt="$(date +%s)" --overwrite
- name: Verify rollout
run: |
kubectl rollout status deployment/hostexecutor \
--namespace hostexecutor \
--timeout=120s \
|| {
kubectl get pods -n hostexecutor -o wide
kubectl get events -n hostexecutor --sort-by='.lastTimestamp' | tail -20
exit 1
}
+17
View File
@@ -34,3 +34,20 @@ bin/
# downloaded + processed market data (track via DVC/MinIO, #10 — not git)
data/
# autoresearch run dirs (ephemeral; each scaffold rebuilds from fixtures/)
runs/
# ephemeral experiment outputs (generated by train.py / loop.py)
metrics.json
embeddings.json
HEARTBEAT
STATUS.md
# python caches
__pycache__/
*.pyc
.pytest_cache/
# built Go binaries
eval
+16 -1
View File
@@ -34,9 +34,24 @@ tasks:
data:prepare:all:
desc: "Build both daily and hourly parquets"
deps: [data:prepare:daily, data:prepare:hourly]
train:multipair:
desc: "Train 5-pair G10 HEPA (D=256, best config, phase1_r2≈0.44)"
cmds: [JEPA_USE_MULTIPAIR=1 JEPA_D_MODEL=256 .venv/bin/python train.py]
data:fetch:multipair:
desc: "Download G10 M1 data (GBPUSD/USDJPY/USDCHF/AUDUSD) 2008-2023 from histdata"
cmds: [.venv/bin/python scripts/fetch_multipair.py]
data:prepare:pair:
desc: "Build {PAIR}_hourly.parquet from data/raw/{PAIR}/ (e.g. PAIR=gbpusd)"
cmds: ["PAIR={{.PAIR}} .venv/bin/python scripts/prepare_hourly.py {{.EXTRA_ARGS}}"]
vars:
PAIR: '{{default "eurusd" .PAIR}}'
data:prepare:multipair:
desc: "Merge 5-pair hourly parquets into eurusd_multipair.parquet"
cmds: [.venv/bin/python scripts/prepare_multipair.py]
data:test:
desc: "Run Python data pipeline tests"
cmds: [.venv/bin/python -m pytest tests/test_prepare_hourly.py tests/test_hepa.py -v]
cmds: [.venv/bin/python -m pytest tests/test_prepare_hourly.py tests/test_hepa.py tests/test_multipair.py -v]
eval:probe:
desc: "Run linear-probe (val_vol_r2) on embeddings from metrics.json"
+26
View File
@@ -83,6 +83,32 @@ func main() {
fmt.Printf(`{"metric":"effective_rank","value":%.6f}`+"\n", er)
log.Info("effective rank", "erank", fmt.Sprintf("%.2f", er))
case "var":
// Parametric 99% VaR breach rate from probe predictions vs actual realized vol.
// Requires train_embeddings (for no-leakage probe fit) and realized_vol (OOS).
if len(d.RealizedVol) == 0 {
log.Error("var requires realized_vol in embeddings.json")
os.Exit(1)
}
var predVol []float64
if len(d.TrainEmbeddings) > 0 {
trEmb, mu, sd := standardiseCompute(d.TrainEmbeddings)
oosEmb := applyStandardise(d.Embeddings, mu, sd)
predVol = eval.LinearProbePredict(trEmb, d.TrainRealizedVol, oosEmb, 1e-3)
} else {
oosEmb, mu, sd := standardiseCompute(d.Embeddings)
n70 := int(float64(len(oosEmb)) * 0.7)
oos70 := applyStandardise(d.Embeddings[n70:], mu, sd)
predVol = eval.LinearProbePredict(oosEmb[:n70], d.RealizedVol[:n70], oos70, 1e-3)
d.RealizedVol = d.RealizedVol[n70:]
}
const z99 = 2.326
breachRate, kupiecP := eval.VaRBreachRate(predVol, d.RealizedVol, z99)
fmt.Printf(`{"metric":"VaR_breach_rate_99_oos_regime_cond","value":%.6f,"kupiec_p":%.6f}`+"\n",
breachRate, kupiecP)
log.Info("VaR breach rate 99%", "breach_rate", fmt.Sprintf("%.4f", breachRate),
"kupiec_p", fmt.Sprintf("%.4f", kupiecP))
default:
log.Error("unknown metric", "metric", *metric)
os.Exit(1)
Binary file not shown.
+11
View File
@@ -0,0 +1,11 @@
{
"strategic_question": "What is the highest-leverage path to a JEPA-based FX tail-risk system that beats a GARCH/EWMA baseline on out-of-sample VaR-breach calibration, given one GPU and a solo researcher?",
"nodes": [
{
"id": "phase-a-toy",
"status": "autoresearch-ready",
"question": "Improve the OOS linear-probe R² (val_vol_r2) of the HEPA encoder on EUR/USD daily realized vol. The encoder is a small causal transformer trained with VICReg. Vary one hyperparameter or architectural choice per iteration — model size, learning rate, window, patch length, depth, VICReg loss weights — to push val_vol_r2 as high as possible on the 2022-2023 OOS slice.",
"candidate_metric": "val_vol_r2"
}
]
}
+107
View File
@@ -0,0 +1,107 @@
package eval
import "math"
// VaRBreachRate computes the parametric 99% VaR breach rate and Kupiec POF p-value.
//
// VaR_99_t = predVol[t] × z99 (z99 = 2.326 for 99% normal VaR)
// breach_t = actualVol[t] > VaR_99_t (strict inequality)
// breachRate = fraction of breaches over all steps
// kupiecP = Kupiec POF p-value: P(chi²(1) > LR) where LR is the likelihood ratio
// testing H0: true breach probability = 1%. High p = well-calibrated.
//
// Returns (0, 1) for empty or mismatched input.
func VaRBreachRate(predVol, actualVol []float64, z99 float64) (breachRate, kupiecP float64) {
n := len(predVol)
if n == 0 || n != len(actualVol) {
return 0, 1
}
var n1 int
for i := 0; i < n; i++ {
if actualVol[i] > predVol[i]*z99 {
n1++
}
}
breachRate = float64(n1) / float64(n)
kupiecP = kupiecPOF(n, n1, 0.01)
return
}
// kupiecPOF returns the Kupiec Proportion-of-Failures p-value.
// H0: true breach probability = p0 (e.g. 0.01 for 99% VaR).
// Returns 1.0 for edge cases (n=0, p_hat=p0).
func kupiecPOF(n, n1 int, p0 float64) float64 {
if n == 0 {
return 1.0
}
n0 := n - n1
phat := float64(n1) / float64(n)
var lr float64
switch n1 {
case 0:
// 0 × ln(0/p0) = 0 by convention; only the n0 term contributes
lr = 2 * float64(n0) * math.Log((1-phat)/(1-p0))
case n:
// n0 term vanishes
lr = 2 * float64(n1) * math.Log(phat/p0)
default:
lr = 2 * (float64(n1)*math.Log(phat/p0) + float64(n0)*math.Log((1-phat)/(1-p0)))
}
if lr <= 0 {
return 1.0
}
// P(chi²(1) > LR) = erfc(sqrt(LR/2)) [chi²(1) = Z², Z~N(0,1)]
return math.Erfc(math.Sqrt(lr / 2))
}
// LinearProbePredict fits ridge regression on (trainEmb, trainY) and returns
// predictions for testEmb. Complements LinearProbeTrainTest when the caller
// needs the raw predictions (e.g. to compute VaR breach rate).
// Returns nil when trainEmb is empty.
func LinearProbePredict(trainEmb [][]float64, trainY []float64,
testEmb [][]float64, lambda float64) []float64 {
n := len(trainEmb)
if n == 0 || len(testEmb) == 0 {
return nil
}
d := len(trainEmb[0])
p := d + 1
A := make([][]float64, n)
for i, e := range trainEmb {
row := make([]float64, p)
copy(row, e)
row[d] = 1.0
A[i] = row
}
AtA := make([][]float64, p)
for i := range AtA {
AtA[i] = make([]float64, p)
}
Aty := make([]float64, p)
for i := 0; i < n; i++ {
for j := 0; j < p; j++ {
Aty[j] += A[i][j] * trainY[i]
for k := 0; k < p; k++ {
AtA[j][k] += A[i][j] * A[i][k]
}
}
}
for j := 0; j < p; j++ {
AtA[j][j] += lambda
}
w := solveCholesky(AtA, Aty)
preds := make([]float64, len(testEmb))
for i, e := range testEmb {
row := make([]float64, p)
copy(row, e)
row[d] = 1.0
preds[i] = dot(row, w)
}
return preds
}
+138
View File
@@ -0,0 +1,138 @@
package eval_test
import (
"math"
"testing"
"gitea.d-ma.be/mathias/jepa-fx-risk/internal/eval"
)
// ── VaRBreachRate golden tests ──────────────────────────────────────────────
//
// VaR_99_t = predVol[t] × z99 (parametric 99% normal VaR)
// breach_t = actualVol[t] > VaR_99_t
// breachRate = mean(breach_t)
// kupiecP = Kupiec POF p-value (chi²(1) test, H0: breach rate = 1%)
func TestVaRBreachRate_ZeroBreaches(t *testing.T) {
// 0.02 < 0.01×2.326=0.02326 → no breaches
pred := []float64{0.01, 0.01, 0.01}
act := []float64{0.02, 0.02, 0.02}
rate, _ := eval.VaRBreachRate(pred, act, 2.326)
if rate != 0 {
t.Fatalf("want rate=0, got %.4f", rate)
}
}
func TestVaRBreachRate_AllBreach(t *testing.T) {
// 0.03 > 0.02326 → all breach
pred := []float64{0.01, 0.01}
act := []float64{0.03, 0.03}
rate, _ := eval.VaRBreachRate(pred, act, 2.326)
if math.Abs(rate-1.0) > 1e-9 {
t.Fatalf("want rate=1.0, got %.4f", rate)
}
}
func TestVaRBreachRate_Golden(t *testing.T) {
// n=10, 2 breaches at indices 0 and 2 → rate=0.2
// Kupiec: p_hat=0.2 vs p0=0.01 → strongly reject H0 (p < 0.05)
pred := make([]float64, 10)
act := make([]float64, 10)
for i := range pred {
pred[i] = 0.01
act[i] = 0.01 // no breach: 0.01 < 0.02326
}
act[0] = 0.03 // breach
act[2] = 0.03 // breach
rate, kupiecP := eval.VaRBreachRate(pred, act, 2.326)
if math.Abs(rate-0.2) > 1e-9 {
t.Fatalf("breach rate: want 0.2, got %.4f", rate)
}
if kupiecP > 0.05 {
t.Fatalf("kupiec p-value: want <0.05 (strong reject H0), got %.4f", kupiecP)
}
}
func TestVaRBreachRate_PerfectCalibration(t *testing.T) {
// n=100, exactly 1 breach → p_hat=0.01=p0 → LR=0 → kupiecP≈1.0
n := 100
pred := make([]float64, n)
act := make([]float64, n)
for i := range pred {
pred[i] = 0.01
act[i] = 0.015 // < 0.02326, no breach
}
act[0] = 0.025 // > 0.02326, breach
rate, kupiecP := eval.VaRBreachRate(pred, act, 2.326)
if math.Abs(rate-0.01) > 1e-9 {
t.Fatalf("breach rate: want 0.01, got %.4f", rate)
}
if kupiecP < 0.9 {
t.Fatalf("kupiec p-value: want ≈1.0 (well calibrated), got %.4f", kupiecP)
}
}
func TestVaRBreachRate_EmptyInput(t *testing.T) {
rate, kupiecP := eval.VaRBreachRate(nil, nil, 2.326)
if rate != 0 || kupiecP != 1 {
t.Fatalf("empty: want (0,1), got (%.4f,%.4f)", rate, kupiecP)
}
}
func TestVaRBreachRate_LenMismatch(t *testing.T) {
rate, kupiecP := eval.VaRBreachRate([]float64{0.01}, []float64{0.01, 0.02}, 2.326)
if rate != 0 || kupiecP != 1 {
t.Fatalf("mismatch: want (0,1), got (%.4f,%.4f)", rate, kupiecP)
}
}
func TestVaRBreachRate_Z99Default(t *testing.T) {
// z99=2.326 is the canonical value; test that boundary case works
// VaR = 0.01 × 2.326 = 0.02326
// actual = 0.02326 → NOT a breach (strict >)
pred := []float64{0.01}
act := []float64{0.02326}
rate, _ := eval.VaRBreachRate(pred, act, 2.326)
if rate != 0 {
t.Fatalf("boundary: exactly at VaR is not a breach; want rate=0, got %.4f", rate)
}
}
// ── LinearProbePredict ──────────────────────────────────────────────────────
func TestLinearProbePredict_PerfectLinear(t *testing.T) {
// y = x; predictions should match targets closely
n := 20
trainEmb := make([][]float64, n)
trainY := make([]float64, n)
testEmb := make([][]float64, 5)
testY := []float64{5, 10, 15, 20, 25}
for i := range trainEmb {
trainEmb[i] = []float64{float64(i)}
trainY[i] = float64(i)
}
for i := range testEmb {
testEmb[i] = []float64{testY[i]}
}
preds := eval.LinearProbePredict(trainEmb, trainY, testEmb, 1e-3)
if len(preds) != len(testEmb) {
t.Fatalf("len: want %d, got %d", len(testEmb), len(preds))
}
for i, p := range preds {
if math.Abs(p-testY[i]) > 1.0 {
t.Fatalf("pred[%d]: want ≈%.1f, got %.4f", i, testY[i], p)
}
}
}
func TestLinearProbePredict_EmptyTrain(t *testing.T) {
preds := eval.LinearProbePredict(nil, nil, [][]float64{{1.0}}, 1e-3)
if len(preds) != 0 {
t.Fatalf("empty train: want nil/empty preds, got len=%d", len(preds))
}
}
+136 -54
View File
@@ -4,7 +4,7 @@ Agent (on iguana/berget — NOT koala, whose GPU is reserved for train.py) reads
program.md + train.py + STATUS.md, proposes ONE change to train.py, we run it,
keep if val_vol_r2 improved else git-revert. Appends per-iter record to STATUS.md.
LITELLM_KEY=xxx python loop.py [--iters N] [--model MODEL]
LITELLM_KEY=xxx python loop.py [--iters N] [--model MODEL] [--run-dir runs/rq-04]
Env:
LITELLM_KEY — LiteLLM master key (required)
@@ -12,6 +12,7 @@ Env:
LOOP_MODEL — default berget/gemma4-31b (non-thinking; iguana/berget only)
LOOP_ITERS — default 3
TRAIN_TIMEOUT — seconds per train.py run, default 120
NTFY_URL — optional: POST crash/stall alerts here (e.g. ntfy.sh/<topic>)
"""
import argparse
import json
@@ -24,14 +25,19 @@ from pathlib import Path
import urllib.request
LITELLM_BASE = os.environ.get("LITELLM_BASE", "http://localhost:30401/v1")
LITELLM_KEY = os.environ.get("LITELLM_KEY", "")
LOOP_MODEL = os.environ.get("LOOP_MODEL", "berget/gemma4-31b")
LOOP_ITERS = int(os.environ.get("LOOP_ITERS", "3"))
LITELLM_BASE = os.environ.get("LITELLM_BASE", "http://localhost:30401/v1")
LITELLM_KEY = os.environ.get("LITELLM_KEY", "")
LOOP_MODEL = os.environ.get("LOOP_MODEL", "berget/gemma4-31b")
LOOP_ITERS = int(os.environ.get("LOOP_ITERS", "3"))
TRAIN_TIMEOUT = int(os.environ.get("TRAIN_TIMEOUT", "120"))
NTFY_URL = os.environ.get("NTFY_URL", "")
# Resolved by main() once --run-dir is parsed.
RUN_DIR = Path(".")
STATUS_MD = Path("STATUS.md")
METRICS_JSON = Path("metrics.json")
TRAIN_PY = Path("train.py")
HEARTBEAT = Path("HEARTBEAT")
AGENT_SYSTEM = textwrap.dedent("""\
You are the autoresearch agent for jepa-fx-risk. Your job: propose ONE small,
@@ -64,7 +70,7 @@ def gpu_snapshot() -> str:
return "gpu=N/A"
def read_metric() -> float | None:
def read_metric() -> "float | None":
if not METRICS_JSON.exists():
return None
try:
@@ -73,14 +79,19 @@ def read_metric() -> float | None:
return None
def run_train() -> tuple[float | None, float, str]:
"""Run train.py. Returns (val_vol_r2 or None, wall_secs, stderr_tail)."""
def run_train() -> "tuple[float | None, float, str]":
"""Run train.py from project root with METRICS_OUT pointing into the run dir."""
t0 = time.time()
gpu_before = gpu_snapshot()
env = dict(os.environ)
env["METRICS_OUT"] = str(METRICS_JSON.resolve())
# train.py is copied into the run dir, so sys.path[0] is that run dir — which
# has no scripts/. Put the project root (where loop.py + scripts/ live) on
# PYTHONPATH so train.py's `from scripts.var_breach import ...` resolves.
env["PYTHONPATH"] = str(Path(__file__).resolve().parent) + os.pathsep + env.get("PYTHONPATH", "")
try:
r = subprocess.run(
[sys.executable, "train.py"],
capture_output=True, text=True, timeout=TRAIN_TIMEOUT,
[sys.executable, str(TRAIN_PY.resolve())],
capture_output=True, text=True, timeout=TRAIN_TIMEOUT, env=env,
)
elapsed = time.time() - t0
if r.returncode != 0:
@@ -91,10 +102,10 @@ def run_train() -> tuple[float | None, float, str]:
return None, TRAIN_TIMEOUT, "TIMEOUT"
def call_agent(iteration: int, best_so_far: float | None) -> str:
def call_agent(iteration: int, best_so_far: "float | None") -> str:
"""Ask the LLM agent to edit train.py. Returns new train.py content."""
context = "\n\n".join([
"# program.md\n" + read_file(Path("program.md")),
"# program.md\n" + read_file(RUN_DIR / "program.md"),
"# train.py (current)\n" + read_file(TRAIN_PY),
"# STATUS.md (history)\n" + read_file(STATUS_MD)[-2000:],
"# metrics.json (last run)\n" + read_file(METRICS_JSON),
@@ -132,73 +143,144 @@ def append_status(line: str):
f.write(line + "\n")
def write_heartbeat(iteration: int, status: str = "alive"):
"""Update HEARTBEAT so watchdogs can detect stalls."""
HEARTBEAT.write_text("%s iter=%d ts=%.0f\n" % (status, iteration, time.time()))
def ntfy(msg: str):
"""POST an alert to NTFY_URL (best-effort; silently ignored on any error)."""
if not NTFY_URL:
return
try:
req = urllib.request.Request(
NTFY_URL, data=msg.encode(), method="POST",
headers={"Content-Type": "text/plain"},
)
urllib.request.urlopen(req, timeout=5)
except Exception:
pass
def main():
global RUN_DIR, STATUS_MD, METRICS_JSON, TRAIN_PY, HEARTBEAT
parser = argparse.ArgumentParser()
parser.add_argument("--iters", type=int, default=LOOP_ITERS)
parser.add_argument("--model", default=LOOP_MODEL)
parser.add_argument(
"--run-dir", default=None,
help="run dir scaffolded by autoresearch_start.py; "
"STATUS.md, metrics.json, HEARTBEAT, and train.py live here",
)
args = parser.parse_args()
loop_iters = args.iters
loop_model = args.model
if args.run_dir:
RUN_DIR = Path(args.run_dir)
if not RUN_DIR.is_dir():
print("ERROR: run dir not found:", RUN_DIR); sys.exit(1)
STATUS_MD = RUN_DIR / "STATUS.md"
METRICS_JSON = RUN_DIR / "metrics.json"
TRAIN_PY = RUN_DIR / "train.py"
HEARTBEAT = RUN_DIR / "HEARTBEAT"
if not LITELLM_KEY:
print("ERROR: set LITELLM_KEY"); sys.exit(1)
if not STATUS_MD.exists():
STATUS_MD.write_text("# Autoresearch STATUS\n\n| iter | val_vol_r2 | delta | action | secs | gpu | change |\n|------|-----------|-------|--------|------|-----|--------|\n")
STATUS_MD.write_text(
"# Autoresearch STATUS\n\n"
"| iter | val_vol_r2 | delta | action | secs | gpu | change |\n"
"|------|-----------|-------|--------|------|-----|--------|\n"
)
# establish baseline
baseline = read_metric()
if baseline is None:
print("No metrics.json — running train.py for baseline...")
m, secs, err = run_train()
if m is None:
print("Baseline run failed:", err); sys.exit(1)
msg = "Baseline run failed: " + err
print(msg)
ntfy("[jepa-fx-risk] loop CRASH — " + msg)
sys.exit(1)
baseline = m
print("Baseline: val_vol_r2 = %.4f (%.1fs)" % (baseline, secs))
best = baseline
print("Starting loop | model=%s | iters=%d | baseline=%.4f" % (LOOP_MODEL, LOOP_ITERS, best))
print("Starting loop | model=%s | iters=%d | baseline=%.4f" % (loop_model, loop_iters, best))
if args.run_dir:
print(" run-dir:", RUN_DIR)
for i in range(1, LOOP_ITERS + 1):
print("\n--- iter %d/%d ---" % (i, LOOP_ITERS))
original = TRAIN_PY.read_text()
iter_index = 0
try:
for i in range(1, loop_iters + 1):
iter_index = i
write_heartbeat(i, "agent-call")
print("\n--- iter %d/%d ---" % (i, loop_iters))
original = TRAIN_PY.read_text()
print(" calling agent (%s)..." % LOOP_MODEL)
t_agent = time.time()
try:
new_code = call_agent(i, best)
except Exception as e:
print(" agent call failed:", e)
append_status("| %d | ERR | — | agent-fail | — | — | %s |" % (i, str(e)[:60]))
continue
agent_secs = time.time() - t_agent
print(" agent replied in %.1fs" % agent_secs)
print(" calling agent (%s)..." % loop_model)
t_agent = time.time()
try:
new_code = call_agent(i, best)
except Exception as e:
msg = str(e)
print(" agent call failed:", msg)
append_status("| %d | ERR | — | agent-fail | — | — | %s |" % (i, msg[:60]))
write_heartbeat(i, "agent-fail")
ntfy("[jepa-fx-risk] iter %d agent FAIL — %s" % (i, msg[:80]))
continue
agent_secs = time.time() - t_agent
print(" agent replied in %.1fs" % agent_secs)
# strip accidental markdown fences
if new_code.strip().startswith("```"):
lines = new_code.strip().splitlines()
new_code = "\n".join(lines[1:-1] if lines[-1].strip() == "```" else lines[1:])
# strip accidental markdown fences
if new_code.strip().startswith("```"):
lines = new_code.strip().splitlines()
new_code = "\n".join(lines[1:-1] if lines[-1].strip() == "```" else lines[1:])
TRAIN_PY.write_text(new_code)
TRAIN_PY.write_text(new_code)
gpu = gpu_snapshot()
print(" running train.py [%s]..." % gpu)
metric, secs, err = run_train()
write_heartbeat(i, "training")
gpu = gpu_snapshot()
print(" running train.py [%s]..." % gpu)
metric, secs, err = run_train()
if metric is None:
print(" train.py FAILED — reverting. err:", err[:100])
revert_train(original)
append_status("| %d | FAIL | — | revert | %.0fs | %s | run error |" % (i, secs, gpu))
continue
if metric is None:
print(" train.py FAILED — reverting. err:", err[:100])
revert_train(original)
append_status("| %d | FAIL | — | revert | %.0fs | %s | run error |" % (i, secs, gpu))
write_heartbeat(i, "train-fail")
ntfy("[jepa-fx-risk] iter %d train FAIL — %s" % (i, err[:80]))
continue
delta = metric - best
if metric > best:
best = metric
action = "KEEP"
else:
revert_train(original)
action = "revert"
delta = metric - best
if metric > best:
best = metric
action = "KEEP"
else:
revert_train(original)
action = "revert"
summary = "| %d | %.4f | %+.4f | %s | %.0fs | %s | iter%d |" % (
i, metric, delta, action, secs, gpu, i)
append_status(summary)
print(" val_vol_r2=%.4f delta=%+.4f action=%s [%.0fs]" % (metric, delta, action, secs))
summary = "| %d | %.4f | %+.4f | %s | %.0fs | %s | iter%d |" % (
i, metric, delta, action, secs, gpu, i)
append_status(summary)
write_heartbeat(i, "done")
print(" val_vol_r2=%.4f delta=%+.4f action=%s [%.0fs]" % (metric, delta, action, secs))
except Exception as e:
msg = "loop CRASH at iter %d: %s" % (iter_index, e)
print("FATAL:", msg)
ntfy("[jepa-fx-risk] " + msg)
raise
print("\nDone. Best val_vol_r2 = %.4f (baseline was %.4f, delta %+.4f)" % (best, baseline, best - baseline))
print("STATUS.md updated.")
write_heartbeat(loop_iters, "done")
ntfy("[jepa-fx-risk] loop done. best val_vol_r2=%.4f (delta %+.4f)" % (best, best - baseline))
if __name__ == "__main__":
-14
View File
@@ -1,14 +0,0 @@
{
"val_vol_r2": 0.3641397896593044,
"phase1_r2": 0.3908407688140869,
"n_test": 11641,
"knobs": {
"WINDOW": 120,
"PATCH_LEN": 24,
"D_MODEL": 128,
"DEPTH": 2,
"ALPHA": 0.1,
"DELTA_T_MAX": 3,
"EPOCHS": 300
}
}
+3 -1
View File
@@ -5,4 +5,6 @@
numpy>=2.0
pandas>=2.2
pyarrow>=16
histdata>=1.3 # histdata.com downloader (handles the tk token politely)
histdata>=1.1 # histdata.com downloader (1.1 is newest on PyPI; 1.3 never existed)
hmmlearn>=0.3 # regime detector (prepare_regime.py, jepa-fx-risk#13)
scikit-learn>=1.4 # HMM dependency
+155
View File
@@ -0,0 +1,155 @@
"""autoresearch start — scaffold a run dir from an Autoresearch Council backlog leaf.
Usage:
python scripts/autoresearch_start.py <backlog.json> <rq-id>
Reads the Council backlog JSON (from agentsquad autoresearch_pipe.py Stage-3 output),
finds the node by rq-id, validates it is autoresearch-ready (fail-closed), then
scaffolds runs/<rq-id>/ with:
program.md — hypothesis, single metric (stripped), agent search-space seam
run.json — provenance (strategic_question + council_node) + config
train.py — copy of project train.py (the loop edits this, keeps history clean)
Launch:
LITELLM_KEY=xxx python loop.py --run-dir runs/<rq-id>
Refs: jepa-fx-risk#11, agentsquad#44
"""
import json
import shutil
import sys
from datetime import datetime, timezone
from pathlib import Path
def load_backlog(path: str) -> dict:
try:
with open(path) as f:
return json.load(f)
except FileNotFoundError:
print(f"error: backlog file not found: {path}", file=sys.stderr)
raise
def scaffold_run(
backlog_path_or_dict,
rq_id: str,
run_dir: Path,
train_py_src: Path,
) -> None:
"""Scaffold a run dir. Raises SystemExit on any validation failure."""
if isinstance(backlog_path_or_dict, (str, Path)):
backlog = load_backlog(str(backlog_path_or_dict))
else:
backlog = backlog_path_or_dict
# Find node
nodes_by_id = {n["id"]: n for n in backlog.get("nodes", [])}
if rq_id not in nodes_by_id:
print(f"error: rq-id {rq_id!r} not found in backlog", file=sys.stderr)
sys.exit(1)
node = nodes_by_id[rq_id]
# Fail-closed: only autoresearch-ready nodes may be scaffolded
status = node.get("status", "")
if status != "autoresearch-ready":
print(
f"error: {rq_id} has status {status!r}, not 'autoresearch-ready' — refusing to scaffold",
file=sys.stderr,
)
sys.exit(1)
# Guard against overwriting an existing run
if run_dir.exists():
print(
f"error: {run_dir} already exists — remove it first to re-scaffold",
file=sys.stderr,
)
sys.exit(1)
metric = (node.get("candidate_metric") or "").strip()
strategic_q = backlog.get("strategic_question", "")
council_node = node["id"]
generated_at = datetime.now(timezone.utc).isoformat()
run_dir.mkdir(parents=True)
# --- program.md ---
program_md = f"""# program.md — {council_node}: {node.get("question", "")[:80]}
## Provenance
- strategic_question: {json.dumps(strategic_q)}
- council_node: {council_node} (autoresearch-ready; Autoresearch Council backlog)
- generated_at: {generated_at}
## Hypothesis
{node.get("question", "")}
## Single validation metric (optimise this, nothing else)
`{metric}` — see eval harness for the exact definition. Only this scalar drives
keep/revert decisions. Report alongside but do NOT optimise:
- Kupiec POF p-value (calibration sanity)
- val_vol_r2 (representation quality guard)
## What the agent MAY modify (the search space)
- Hyperparameters in train.py (model size, LR, window, patch_len, epochs, etc.)
- Conditioning mechanisms (e.g. JEPA_ENABLE_REGIME toggle)
- Loss function weights and architecture depth
## Frozen (do NOT touch — keeps the ablation clean)
- Data pipeline and splits (train ≤2021, OOS ≥2022, test 2024 held out)
- The metric definition and scoring code
- loop.py, scripts/, tests/
## Experiment loop (per Karpathy autoresearch)
Each iter (≤ time-box): apply ONE change to train.py → run → read
`{metric}` → keep if improved (and Kupiec p-value did not collapse), else revert.
Stop on: target reached, max iters, or K consecutive iters with no improvement.
"""
(run_dir / "program.md").write_text(program_md)
# --- run.json (provenance + config) ---
run_meta = {
"strategic_question": strategic_q,
"council_node": council_node,
"metric": metric,
"generated_at": generated_at,
"model_tier": "homelab",
"max_iters": 10,
"time_box_minutes": 5,
}
(run_dir / "run.json").write_text(json.dumps(run_meta, indent=2) + "\n")
# --- train.py (loop edits this copy; project root train.py is the template) ---
shutil.copy(train_py_src, run_dir / "train.py")
def main() -> None:
if len(sys.argv) != 3:
print("usage: python scripts/autoresearch_start.py <backlog.json> <rq-id>")
sys.exit(1)
backlog_path, rq_id = sys.argv[1], sys.argv[2]
project_root = Path(__file__).parent.parent
run_dir = project_root / "runs" / rq_id
train_py_src = project_root / "train.py"
scaffold_run(backlog_path, rq_id, run_dir, train_py_src)
backlog = load_backlog(backlog_path)
nodes_by_id = {n["id"]: n for n in backlog.get("nodes", [])}
metric = (nodes_by_id[rq_id].get("candidate_metric") or "").strip()
print(f"✓ scaffolded {run_dir}")
print(f" node: {rq_id}")
print(f" metric: {metric}")
print()
print("launch:")
print(f" LITELLM_KEY=xxx python loop.py --run-dir runs/{rq_id}")
if __name__ == "__main__":
main()
+48
View File
@@ -0,0 +1,48 @@
"""Fetch G10 FX M1 data from histdata.com for all pairs except EURUSD (already fetched).
Each pair's zips go into data/raw/{pair}/ to avoid collisions.
Output: data/raw/gbpusd/DAT_ASCII_GBPUSD_M1_YYYY.zip etc.
python scripts/fetch_multipair.py
PAIRS=gbpusd,usdjpy YEARS=2020,2021 python scripts/fetch_multipair.py
"""
import os
import time
from histdata import download_hist_data
from histdata.api import Platform as P, TimeFrame as T
PAIRS_DEFAULT = ["gbpusd", "usdjpy", "usdchf", "audusd"]
YEARS_DEFAULT = list(range(2008, 2024))
def main():
pairs_env = os.environ.get("PAIRS", "")
pairs = [p.strip() for p in pairs_env.split(",")] if pairs_env else PAIRS_DEFAULT
years_env = os.environ.get("YEARS", "")
years = [int(y.strip()) for y in years_env.split(",")] if years_env else YEARS_DEFAULT
for pair in pairs:
out_dir = f"data/raw/{pair}"
os.makedirs(out_dir, exist_ok=True)
print(f"\n=== {pair.upper()} ===")
for yr in years:
out_path = os.path.join(out_dir, f"DAT_ASCII_{pair.upper()}_M1_{yr}.zip")
if os.path.exists(out_path):
print(f" {yr} already present, skip")
continue
try:
f = download_hist_data(
year=str(yr), month=None, pair=pair,
platform=P.GENERIC_ASCII, time_frame=T.ONE_MINUTE,
output_directory=out_dir,
)
print(f" fetched {yr}{f}")
except Exception as e:
print(f" {yr} FAILED: {e}")
time.sleep(2)
if __name__ == "__main__":
main()
+5 -3
View File
@@ -18,8 +18,9 @@ import zipfile
import numpy as np
import pandas as pd
PAIR = os.environ.get("PAIR", "EURUSD").upper()
RAW_DEFAULT = "data/raw"
OUT_DEFAULT = "data/processed/eurusd_hourly.parquet"
OUT_DEFAULT = f"data/processed/{PAIR.lower()}_hourly.parquet"
MIN_BARS = 30 # drop hours thinner than this (holidays, DST boundary artefacts)
@@ -70,9 +71,10 @@ def resample_to_hourly(m1: pd.DataFrame) -> pd.DataFrame:
return agg[cols]
def load_m1_from_zips(raw_dir: str) -> pd.DataFrame:
def load_m1_from_zips(raw_dir: str, pair: str = None) -> pd.DataFrame:
"""Load and concatenate all M1 zips from raw_dir (histdata format)."""
pattern = os.path.join(raw_dir, "DAT_ASCII_EURUSD_M1_*.zip")
p = (pair or PAIR).upper()
pattern = os.path.join(raw_dir, f"DAT_ASCII_{p}_M1_*.zip")
zips = sorted(glob.glob(pattern))
if not zips:
raise FileNotFoundError(f"No M1 zips found at {pattern}")
+72
View File
@@ -0,0 +1,72 @@
"""Merge per-pair hourly parquets into a single wide multipair parquet.
Each pair contributes two features: {pair}_ret and {pair}_rv (realized vol).
The merge is an INNER JOIN on datetime — only hours present in ALL pairs are kept.
The target for train.py remains eurusd_rv.
Output: data/processed/eurusd_multipair.parquet
python scripts/prepare_multipair.py
PROCESSED=data/processed python scripts/prepare_multipair.py
"""
import os
import pandas as pd
PAIRS = ["eurusd", "gbpusd", "usdjpy", "usdchf", "audusd"]
PROCESSED_DEFAULT = "data/processed"
OUT_DEFAULT = "data/processed/eurusd_multipair.parquet"
def merge_pair_parquets(pair_dfs: dict) -> pd.DataFrame:
"""Inner-join hourly DataFrames from multiple pairs on datetime.
Args:
pair_dfs: dict mapping pair name (e.g. "eurusd") to hourly DataFrame
with columns [datetime, close, ret, realized_vol, ...].
Returns:
Wide DataFrame with columns:
datetime, {pair}_ret, {pair}_rv for each pair.
"""
merged = None
for pair, df in pair_dfs.items():
sub = df[["datetime", "ret", "realized_vol"]].copy()
sub = sub.rename(columns={"ret": f"{pair}_ret", "realized_vol": f"{pair}_rv"})
sub = sub.set_index("datetime")
if merged is None:
merged = sub
else:
merged = merged.join(sub, how="inner")
return merged.reset_index()
def build_multipair_parquet(
processed_dir: str = PROCESSED_DEFAULT,
out_path: str = OUT_DEFAULT,
pairs: list = None,
) -> None:
if pairs is None:
pairs = PAIRS
pair_dfs = {}
for pair in pairs:
path = os.path.join(processed_dir, f"{pair}_hourly.parquet")
if not os.path.exists(path):
raise FileNotFoundError(
f"{pair}_hourly.parquet not found at {path} — run prepare_hourly.py for this pair first"
)
df = pd.read_parquet(path)
pair_dfs[pair] = df
merged = merge_pair_parquets(pair_dfs)
merged.to_parquet(out_path, index=False)
n_pairs = len(pairs)
n_ch = n_pairs * 2
print(f"Multipair parquet: {len(merged):,} rows × {n_ch} feature channels ({n_pairs} pairs)")
print(f"Date range: {merged['datetime'].min()}{merged['datetime'].max()}")
print(f"Written: {out_path}")
if __name__ == "__main__":
processed_dir = os.environ.get("PROCESSED", PROCESSED_DEFAULT)
build_multipair_parquet(processed_dir=processed_dir)
+134
View File
@@ -0,0 +1,134 @@
"""HMM regime detector — 3-state Gaussian HMM on realized_vol.
Fits on the FULL dataset (training + OOS) so the state sequence is globally
consistent across all periods. States are sorted by mean realized vol (ascending):
0 = calm, 1 = stressed, 2 = crisis
Output: data/processed/eurusd_regime.parquet
Columns: datetime (or date), regime (int: 0/1/2)
Deterministic: fixed random_state=42 throughout.
Cached: if the parquet already exists, it is not re-computed.
Usage:
python scripts/prepare_regime.py [--hourly] [--daily] [--force]
jepa-fx-risk#13
"""
import argparse
import os
from pathlib import Path
import numpy as np
import pandas as pd
from hmmlearn import hmm
DATA_DIR = Path(__file__).parent.parent / "data" / "processed"
HOURLY_PATH = DATA_DIR / "eurusd_hourly.parquet"
DAILY_PATH = DATA_DIR / "eurusd_daily.parquet"
OUTPUT_PATH = DATA_DIR / "eurusd_regime.parquet"
N_STATES = 3
RANDOM_STATE = 42
def fit_regime_hmm(realized_vol: np.ndarray, n_states: int = 3, random_state: int = 42) -> np.ndarray:
"""Fit a Gaussian HMM on realized_vol and return state labels (0=calm → n_states-1=crisis).
States are sorted by mean realized vol ascending so label 0 is always calm,
label n_states-1 is always crisis. This makes the labelling deterministic
across datasets with different vol levels.
Args:
realized_vol: 1-D array of realized vol values
n_states: number of HMM hidden states (default 3)
random_state: random seed for reproducibility
Returns:
Integer label array of shape (len(realized_vol),), dtype int64
"""
X = realized_vol.reshape(-1, 1).astype(np.float64)
model = hmm.GaussianHMM(
n_components=n_states,
covariance_type="diag",
min_covar=1e-6,
n_iter=100,
random_state=random_state,
tol=1e-4,
)
model.fit(X)
raw_labels = model.predict(X)
# Sort states by mean realized vol (ascending: calm=0, crisis=n_states-1)
state_means = np.array([X[raw_labels == s].mean() if (raw_labels == s).any() else 0.0
for s in range(n_states)])
rank = np.argsort(state_means) # rank[0] = original state id of the calmest cluster
remap = np.empty(n_states, dtype=np.int64)
for new_label, old_label in enumerate(rank):
remap[old_label] = new_label
return remap[raw_labels].astype(np.int64)
def prepare_regime_df(parquet_path: str, freq: str = "hourly") -> pd.DataFrame:
"""Load parquet, fit HMM, return DataFrame with timestamp + regime columns.
Args:
parquet_path: path to input parquet (hourly or daily)
freq: "hourly" | "daily" — determines timestamp column name
Returns:
DataFrame with columns: (datetime|date), regime
"""
df = pd.read_parquet(parquet_path)
if freq == "hourly":
ts = pd.to_datetime(df["datetime"])
else:
ts = pd.to_datetime(df["date"])
rv = df["realized_vol"].to_numpy(np.float32)
labels = fit_regime_hmm(rv, n_states=N_STATES, random_state=RANDOM_STATE)
return pd.DataFrame({"datetime": ts.values, "regime": labels})
def main():
parser = argparse.ArgumentParser(description="Fit HMM regime detector")
parser.add_argument("--hourly", action="store_true", default=True,
help="use hourly parquet (default)")
parser.add_argument("--daily", action="store_true", default=False,
help="use daily parquet instead of hourly")
parser.add_argument("--force", action="store_true", default=False,
help="overwrite existing output")
parser.add_argument("--out", default=str(OUTPUT_PATH),
help="output parquet path")
args = parser.parse_args()
out_path = Path(args.out)
if out_path.exists() and not args.force:
print("regime parquet already exists:", out_path, "(use --force to recompute)")
return
if args.daily and DAILY_PATH.exists():
src, freq = str(DAILY_PATH), "daily"
elif HOURLY_PATH.exists():
src, freq = str(HOURLY_PATH), "hourly"
elif DAILY_PATH.exists():
src, freq = str(DAILY_PATH), "daily"
else:
raise FileNotFoundError("no parquet found in data/processed/")
print(f"fitting HMM ({N_STATES} states) on {src} ...")
df = prepare_regime_df(src, freq=freq)
counts = df["regime"].value_counts().sort_index()
print("regime distribution:")
for state, count in counts.items():
label = {0: "calm", 1: "stressed", 2: "crisis"}.get(state, f"state{state}")
print(f" {state} ({label}): {count} ({100*count/len(df):.1f}%)")
df.to_parquet(out_path, index=False)
print("wrote:", out_path)
if __name__ == "__main__":
main()
+67
View File
@@ -0,0 +1,67 @@
"""Parametric 99% VaR breach rate + Kupiec POF p-value.
Used by train.py's LOCKED VaR EVAL BLOCK to write VaR_breach_rate_99_oos_regime_cond
to metrics.json so the autoresearch loop can optimise it.
jepa-fx-risk#12
"""
import math
# Canonical metric key — no surrounding whitespace, as required by the loop contract.
METRIC_KEY = "VaR_breach_rate_99_oos_regime_cond"
# Default normal 99th-percentile z-score.
Z99 = 2.326
def var_breach_rate(pred_vol, actual_vol, z99=Z99):
"""Compute VaR breach rate and Kupiec POF p-value.
Args:
pred_vol: iterable of predicted conditional vol forecasts
actual_vol: iterable of actual realized vol (same length)
z99: 99th-percentile z-score (default 2.326)
Returns:
(breach_rate, kupiec_p) where:
breach_rate — fraction of steps where actual_vol > pred_vol × z99
kupiec_p — Kupiec POF p-value (H0: true breach rate = 1%)
High p-value = well-calibrated; low = miscalibrated tail.
"""
pred_v = list(pred_vol)
act_v = list(actual_vol)
n = len(pred_v)
if n == 0 or n != len(act_v):
return 0.0, 1.0
n1 = sum(1 for p, a in zip(pred_v, act_v) if a > p * z99)
breach_rate = n1 / n
p = kupiec_pvalue(n, n1)
return breach_rate, p
def kupiec_pvalue(n, n1, p0=0.01):
"""Kupiec Proportion-of-Failures likelihood ratio test.
H0: true breach probability = p0.
Returns P(chi²(1) > LR) using the identity P(chi²(1)>x) = erfc(sqrt(x/2)).
Returns 1.0 for n=0 or LR<=0 (well-calibrated / over-conservative).
"""
if n == 0:
return 1.0
n0 = n - n1
phat = n1 / n
if n1 == 0:
# 0 × ln(0/p0) = 0 by convention; only n0 term contributes
lr = 2 * n0 * math.log((1 - phat) / (1 - p0))
elif n1 == n:
lr = 2 * n1 * math.log(phat / p0)
else:
lr = 2 * (n1 * math.log(phat / p0) + n0 * math.log((1 - phat) / (1 - p0)))
if lr <= 0:
return 1.0
# P(chi²(1) > LR) = erfc(sqrt(LR/2))
return math.erfc(math.sqrt(lr / 2))
+215
View File
@@ -0,0 +1,215 @@
"""Tests for scripts/autoresearch_start.py — jepa-fx-risk#11 Phase A scaffold.
Success criterion: `autoresearch start <backlog.json> <rq-id>` scaffolds a
runnable run dir from a ready leaf; refuses non-ready nodes; strips
candidate_metric; records provenance.
"""
import importlib.util
import json
import sys
from pathlib import Path
import pytest
# Load the module without executing main()
_SCRIPT = Path(__file__).parent.parent / "scripts" / "autoresearch_start.py"
def _import():
spec = importlib.util.spec_from_file_location("autoresearch_start", _SCRIPT)
mod = importlib.util.module_from_spec(spec)
spec.loader.exec_module(mod)
return mod
@pytest.fixture()
def mod():
return _import()
@pytest.fixture()
def backlog(tmp_path):
data = {
"strategic_question": "Test strategic question?",
"generated_at": "2026-06-27T00:00:00Z",
"nodes": [
{
"id": "rq-01",
"question": "Does X improve Y?",
"case_type": "autoresearch-loop",
"data": "obtainable",
"method": "adjacent",
"falsifiable": "yes",
"candidate_metric": " val_vol_r2", # leading space — bypass test
"depends_on": [],
"status": "autoresearch-ready",
"track": "autoresearch",
"converged": True,
"survived_review": True,
},
{
"id": "rq-02",
"question": "Not ready yet?",
"case_type": "empirical-study",
"data": "obtainable",
"method": "adjacent",
"falsifiable": "yes",
"candidate_metric": None,
"depends_on": [],
"status": "needs-metric",
"track": "study",
"converged": True,
"survived_review": True,
},
{
"id": "rq-03",
"question": "A spike.",
"case_type": "spike",
"data": "have",
"method": "yes-named",
"falsifiable": "yes",
"candidate_metric": None,
"depends_on": [],
"status": "spike-ready",
"track": "spike",
"converged": True,
"survived_review": True,
},
],
}
p = tmp_path / "backlog.json"
p.write_text(json.dumps(data))
return p
@pytest.fixture()
def fake_train_py(tmp_path):
"""Minimal train.py placeholder for scaffold tests."""
src = tmp_path / "train_template.py"
src.write_text("# train.py placeholder\n")
return src
# ---------------------------------------------------------------------------
# fail-closed: refuse non-autoresearch-ready nodes
# ---------------------------------------------------------------------------
class TestRefuseNonReady:
def test_refuses_needs_metric(self, mod, backlog, fake_train_py, tmp_path):
run_dir = tmp_path / "runs" / "rq-02"
with pytest.raises(SystemExit) as exc:
mod.scaffold_run(backlog, "rq-02", run_dir, fake_train_py)
assert exc.value.code != 0
def test_refuses_spike_ready(self, mod, backlog, fake_train_py, tmp_path):
run_dir = tmp_path / "runs" / "rq-03"
with pytest.raises(SystemExit) as exc:
mod.scaffold_run(backlog, "rq-03", run_dir, fake_train_py)
assert exc.value.code != 0
def test_refuses_missing_rq_id(self, mod, backlog, fake_train_py, tmp_path):
run_dir = tmp_path / "runs" / "rq-99"
with pytest.raises(SystemExit) as exc:
mod.scaffold_run(backlog, "rq-99", run_dir, fake_train_py)
assert exc.value.code != 0
def test_refuses_existing_run_dir(self, mod, backlog, fake_train_py, tmp_path):
run_dir = tmp_path / "runs" / "rq-01"
run_dir.mkdir(parents=True)
with pytest.raises(SystemExit) as exc:
mod.scaffold_run(backlog, "rq-01", run_dir, fake_train_py)
assert exc.value.code != 0
# ---------------------------------------------------------------------------
# scaffold structure: correct files created
# ---------------------------------------------------------------------------
class TestScaffoldStructure:
@pytest.fixture(autouse=True)
def _scaffold(self, mod, backlog, fake_train_py, tmp_path):
self.run_dir = tmp_path / "runs" / "rq-01"
mod.scaffold_run(backlog, "rq-01", self.run_dir, fake_train_py)
def test_run_dir_created(self):
assert self.run_dir.is_dir()
def test_program_md_created(self):
assert (self.run_dir / "program.md").exists()
def test_run_json_created(self):
assert (self.run_dir / "run.json").exists()
def test_train_py_copied(self):
assert (self.run_dir / "train.py").exists()
assert (self.run_dir / "train.py").read_text() == "# train.py placeholder\n"
# ---------------------------------------------------------------------------
# program.md content
# ---------------------------------------------------------------------------
class TestProgramMd:
@pytest.fixture(autouse=True)
def _scaffold(self, mod, backlog, fake_train_py, tmp_path):
self.run_dir = tmp_path / "runs" / "rq-01"
mod.scaffold_run(backlog, "rq-01", self.run_dir, fake_train_py)
self.content = (self.run_dir / "program.md").read_text()
def test_contains_hypothesis(self):
assert "Does X improve Y?" in self.content
def test_metric_key_stripped(self):
# candidate_metric had leading space " val_vol_r2" — must be stripped
assert "`val_vol_r2`" in self.content
assert "` val_vol_r2`" not in self.content
def test_contains_strategic_question(self):
assert "Test strategic question?" in self.content
def test_contains_council_node(self):
assert "rq-01" in self.content
# ---------------------------------------------------------------------------
# run.json provenance
# ---------------------------------------------------------------------------
class TestRunJson:
@pytest.fixture(autouse=True)
def _scaffold(self, mod, backlog, fake_train_py, tmp_path):
self.run_dir = tmp_path / "runs" / "rq-01"
mod.scaffold_run(backlog, "rq-01", self.run_dir, fake_train_py)
self.run = json.loads((self.run_dir / "run.json").read_text())
def test_strategic_question_in_provenance(self):
assert self.run["strategic_question"] == "Test strategic question?"
def test_council_node_in_provenance(self):
assert self.run["council_node"] == "rq-01"
def test_metric_stripped_in_provenance(self):
assert self.run["metric"] == "val_vol_r2"
assert self.run["metric"] == self.run["metric"].strip()
def test_generated_at_present(self):
assert "generated_at" in self.run
def test_max_iters_present(self):
assert "max_iters" in self.run
# ---------------------------------------------------------------------------
# load_backlog helper
# ---------------------------------------------------------------------------
class TestLoadBacklog:
def test_loads_json(self, mod, backlog):
data = mod.load_backlog(str(backlog))
assert data["strategic_question"] == "Test strategic question?"
assert len(data["nodes"]) == 3
def test_missing_file_raises(self, mod, tmp_path):
with pytest.raises((FileNotFoundError, SystemExit)):
mod.load_backlog(str(tmp_path / "nonexistent.json"))
+126
View File
@@ -0,0 +1,126 @@
"""Tests for multi-pair G10 pipeline (Option C).
Tests the prepare_multipair.py merge logic and train.py multipair build().
Run: cd ~/dev/AI/jepa-fx-risk && .venv/bin/python -m pytest tests/test_multipair.py -v
"""
import importlib.util
import numpy as np
import pandas as pd
import pytest
import os
def _import_mp():
spec = importlib.util.spec_from_file_location("prepare_multipair", "scripts/prepare_multipair.py")
mod = importlib.util.module_from_spec(spec)
spec.loader.exec_module(mod)
return mod
@pytest.fixture(scope="module")
def mp():
return _import_mp()
def _pair_df(start: str, n_hours: int, seed: int) -> pd.DataFrame:
"""Synthetic single-pair hourly parquet (same schema as prepare_hourly output)."""
rng = np.random.default_rng(seed)
dts = pd.date_range(start, periods=n_hours, freq="h")
closes = 1.1 + np.cumsum(rng.normal(0, 0.001, n_hours))
return pd.DataFrame({
"datetime": dts,
"close": closes,
"ret": rng.normal(0, 0.001, n_hours),
"realized_vol": np.abs(rng.normal(0.0005, 0.0001, n_hours)),
})
# 1. merge_pair_parquets returns inner join on datetime
def test_merge_inner_join(mp):
eur = _pair_df("2020-01-01 00:00", 100, seed=1) # t0 to t0+99h
gbp = _pair_df("2020-01-01 20:00", 60, seed=2) # t0+20 to t0+79h → 60 common
result = mp.merge_pair_parquets({"eurusd": eur, "gbpusd": gbp})
assert len(result) == 60, f"expected 60 (inner join), got {len(result)}"
# 2. merge_pair_parquets prefixes columns with pair name
def test_merge_column_prefixes(mp):
eur = _pair_df("2020-01-01 00:00", 50, seed=1)
gbp = _pair_df("2020-01-01 00:00", 50, seed=2)
result = mp.merge_pair_parquets({"eurusd": eur, "gbpusd": gbp})
assert "datetime" in result.columns, "datetime column missing"
assert "eurusd_ret" in result.columns
assert "eurusd_rv" in result.columns
assert "gbpusd_ret" in result.columns
assert "gbpusd_rv" in result.columns
# raw pair columns should not leak through unprefixed
assert "ret" not in result.columns
assert "realized_vol" not in result.columns
# 3. No NaN in merged output
def test_merge_no_nan(mp):
eur = _pair_df("2020-01-01 00:00", 50, seed=1)
gbp = _pair_df("2020-01-01 00:00", 50, seed=2)
result = mp.merge_pair_parquets({"eurusd": eur, "gbpusd": gbp})
nan_count = result.isnull().sum().sum()
assert nan_count == 0, f"{nan_count} NaN values in merged output"
# 4. PAIRS constant is a non-empty list starting with eurusd
def test_pairs_constant(mp):
assert hasattr(mp, "PAIRS"), "PAIRS constant missing from prepare_multipair.py"
assert len(mp.PAIRS) >= 2, "PAIRS must have at least 2 pairs"
assert mp.PAIRS[0] == "eurusd", "first pair must be eurusd (target pair)"
# 5. merge target column is eurusd_rv (for build() target selection)
def test_merge_has_eurusd_rv_as_target(mp):
eur = _pair_df("2020-01-01 00:00", 50, seed=1)
gbp = _pair_df("2020-01-01 00:00", 50, seed=2)
result = mp.merge_pair_parquets({"eurusd": eur, "gbpusd": gbp})
assert "eurusd_rv" in result.columns, "eurusd_rv (target) missing from merged output"
assert (result["eurusd_rv"] > 0).all(), "eurusd_rv should be positive"
# 6. train.py recognises JEPA_USE_MULTIPAIR env var
def test_use_multipair_knob():
import importlib.util as ilu
spec = ilu.spec_from_file_location(f"train_mp_{id(None)}", "train.py")
mod = ilu.module_from_spec(spec)
saved = os.environ.get("JEPA_USE_MULTIPAIR")
os.environ["JEPA_USE_MULTIPAIR"] = "1"
try:
spec.loader.exec_module(mod)
finally:
if saved is None:
os.environ.pop("JEPA_USE_MULTIPAIR", None)
else:
os.environ["JEPA_USE_MULTIPAIR"] = saved
assert hasattr(mod, "USE_MULTIPAIR"), "USE_MULTIPAIR knob missing from train.py"
assert mod.USE_MULTIPAIR is True
# 7. build() uses n_pairs*2 channels when multipair parquet present
def test_build_uses_multipair_channels():
import importlib.util as ilu
multipair_path = "data/processed/eurusd_multipair.parquet"
if not os.path.exists(multipair_path):
pytest.skip("eurusd_multipair.parquet not present — run data:prepare:multipair first")
saved = os.environ.get("JEPA_USE_MULTIPAIR")
os.environ["JEPA_USE_MULTIPAIR"] = "1"
try:
spec = ilu.spec_from_file_location(f"train_mp2_{id(None)}", "train.py")
mod = ilu.module_from_spec(spec)
spec.loader.exec_module(mod)
(Xtr, _), _ = mod.build()
finally:
if saved is None:
os.environ.pop("JEPA_USE_MULTIPAIR", None)
else:
os.environ["JEPA_USE_MULTIPAIR"] = saved
mp = _import_mp()
expected_ch = len(mp.PAIRS) * 2
assert Xtr.shape[2] == expected_ch, (
f"expected {expected_ch} channels (n_pairs={len(mp.PAIRS)}×2), got {Xtr.shape[2]}"
)
+134
View File
@@ -0,0 +1,134 @@
"""Tests for scripts/prepare_regime.py — HMM regime detector (jepa-fx-risk#13).
TDD: tests first, implementation follows.
"""
import importlib.util
import os
import shutil
from pathlib import Path
import numpy as np
import pandas as pd
import pytest
_SCRIPT = Path(__file__).parent.parent / "scripts" / "prepare_regime.py"
DATA_DIR = Path(__file__).parent.parent / "data" / "processed"
HOURLY = DATA_DIR / "eurusd_hourly.parquet"
DAILY = DATA_DIR / "eurusd_daily.parquet"
def _import():
spec = importlib.util.spec_from_file_location("prepare_regime", _SCRIPT)
mod = importlib.util.module_from_spec(spec)
spec.loader.exec_module(mod)
return mod
@pytest.fixture()
def mod():
return _import()
# ---------------------------------------------------------------------------
# fit_regime_hmm — pure function (doesn't touch disk)
# ---------------------------------------------------------------------------
def _synthetic_rv(seed=42, size=500):
"""Noisy 3-regime vol series: calm→stressed→crisis→calm interleaved."""
rng = np.random.default_rng(seed)
low = np.abs(rng.normal(0.005, 0.001, size=size // 3))
mid = np.abs(rng.normal(0.015, 0.003, size=size // 3))
high = np.abs(rng.normal(0.04, 0.008, size=size - 2 * (size // 3)))
return np.concatenate([low, mid, high])
class TestFitRegimeHmm:
def test_returns_integer_labels(self, mod):
rv = _synthetic_rv(seed=0)
labels = mod.fit_regime_hmm(rv, n_states=3, random_state=42)
assert np.issubdtype(labels.dtype, np.integer), f"dtype={labels.dtype}"
assert len(labels) == len(rv)
def test_states_are_0_1_2(self, mod):
rv = _synthetic_rv(seed=1)
labels = mod.fit_regime_hmm(rv, n_states=3, random_state=42)
unique = set(labels.tolist())
assert unique.issubset({0, 1, 2}), f"unexpected states: {unique}"
def test_deterministic(self, mod):
rv = _synthetic_rv(seed=7)
a = mod.fit_regime_hmm(rv, n_states=3, random_state=42)
b = mod.fit_regime_hmm(rv, n_states=3, random_state=42)
assert np.array_equal(a, b), "HMM not deterministic with same random_state"
def test_sorted_by_vol_asc(self, mod):
# 3 clearly separated noisy clusters; state 0 should be calm, 2 should be crisis.
rng = np.random.default_rng(42)
n = 200
low = np.abs(rng.normal(0.005, 0.001, n))
mid = np.abs(rng.normal(0.015, 0.003, n))
high = np.abs(rng.normal(0.05, 0.008, n))
rv = np.concatenate([low, mid, high])
labels = mod.fit_regime_hmm(rv, n_states=3, random_state=42)
# Mean regime label in the high-vol section should exceed mean in the low-vol section.
assert labels[2*n:].mean() > labels[:n].mean(), \
"crisis section mean regime label should exceed calm section"
# The calm section should not be labeled as crisis (2) dominantly.
calm_modal = int(np.bincount(labels[:n]).argmax())
assert calm_modal < 2, f"calm section mostly labeled {calm_modal}, expected 0 or 1"
def test_two_states(self, mod):
rv = _synthetic_rv(seed=0)
labels = mod.fit_regime_hmm(rv, n_states=2, random_state=42)
unique = set(labels.tolist())
assert unique.issubset({0, 1})
# ---------------------------------------------------------------------------
# prepare_regime_df — reads parquet, fits HMM, returns DataFrame
# ---------------------------------------------------------------------------
class TestPrepareRegimeDf:
@pytest.mark.skipif(not HOURLY.exists(), reason="hourly parquet not available")
def test_output_columns(self, mod):
df = mod.prepare_regime_df(str(HOURLY), freq="hourly")
assert "datetime" in df.columns
assert "regime" in df.columns
@pytest.mark.skipif(not HOURLY.exists(), reason="hourly parquet not available")
def test_regime_values(self, mod):
df = mod.prepare_regime_df(str(HOURLY), freq="hourly")
unique = set(df["regime"].tolist())
assert unique.issubset({0, 1, 2}), f"unexpected regime values: {unique}"
@pytest.mark.skipif(not HOURLY.exists(), reason="hourly parquet not available")
def test_no_nulls(self, mod):
df = mod.prepare_regime_df(str(HOURLY), freq="hourly")
assert df["regime"].isna().sum() == 0
@pytest.mark.skipif(not DAILY.exists(), reason="daily parquet not available")
def test_daily_fallback(self, mod):
df = mod.prepare_regime_df(str(DAILY), freq="daily")
assert "regime" in df.columns
assert set(df["regime"].tolist()).issubset({0, 1, 2})
# ---------------------------------------------------------------------------
# Integration: check that train.py REGIME SEAM exists and is togglable
# ---------------------------------------------------------------------------
class TestTrainPyRegimeSeam:
def test_enable_regime_env_var_documented(self):
train_py = Path(__file__).parent.parent / "train.py"
content = train_py.read_text()
assert "JEPA_ENABLE_REGIME" in content, "JEPA_ENABLE_REGIME toggle not found in train.py"
def test_regime_seam_comment_present(self):
train_py = Path(__file__).parent.parent / "train.py"
content = train_py.read_text()
assert "REGIME" in content and "seam" in content.lower(), \
"agent-editable regime seam marker not found in train.py"
+108
View File
@@ -0,0 +1,108 @@
"""Tests for scripts/var_breach.py — VaR breach rate + Kupiec POF (jepa-fx-risk#12).
Golden tests first: verify the math before wiring it into train.py.
"""
import importlib.util
import math
from pathlib import Path
import pytest
_SCRIPT = Path(__file__).parent.parent / "scripts" / "var_breach.py"
def _import():
spec = importlib.util.spec_from_file_location("var_breach", _SCRIPT)
mod = importlib.util.module_from_spec(spec)
spec.loader.exec_module(mod)
return mod
@pytest.fixture()
def mod():
return _import()
# ---------------------------------------------------------------------------
# var_breach_rate
# ---------------------------------------------------------------------------
class TestVarBreachRate:
def test_zero_breaches(self, mod):
# 0.02 < 0.01×2.326=0.02326 → no breach
rate, _ = mod.var_breach_rate([0.01, 0.01], [0.02, 0.02])
assert rate == 0.0
def test_all_breach(self, mod):
# 0.03 > 0.02326 → all breach
rate, _ = mod.var_breach_rate([0.01, 0.01], [0.03, 0.03])
assert rate == 1.0
def test_golden_two_of_ten(self, mod):
pred = [0.01] * 10
actual = [0.01] * 10
actual[0] = 0.03 # breach
actual[2] = 0.03 # breach
rate, kupiec_p = mod.var_breach_rate(pred, actual)
assert abs(rate - 0.2) < 1e-9, f"rate={rate}"
assert kupiec_p < 0.05, f"kupiec_p={kupiec_p}" # strong reject
def test_perfect_calibration(self, mod):
# n=100, 1 breach → p_hat=0.01=p0=0.01 → LR=0 → kupiec_p≈1
pred = [0.01] * 100
actual = [0.015] * 100
actual[0] = 0.025 # 0.025 > 0.02326 → breach
rate, kupiec_p = mod.var_breach_rate(pred, actual)
assert abs(rate - 0.01) < 1e-9
assert kupiec_p > 0.9, f"kupiec_p={kupiec_p}"
def test_boundary_at_var_is_not_breach(self, mod):
# exactly at VaR_99 is NOT a breach (strict >)
z99 = 2.326
var = 0.01 * z99
rate, _ = mod.var_breach_rate([0.01], [var], z99=z99)
assert rate == 0.0
def test_empty_returns_zero_one(self, mod):
rate, kupiec_p = mod.var_breach_rate([], [])
assert rate == 0.0
assert kupiec_p == 1.0
def test_metric_key_no_whitespace(self, mod):
key = mod.METRIC_KEY
assert key == key.strip(), f"metric key has surrounding whitespace: {key!r}"
assert " " not in key, f"metric key contains space: {key!r}"
def test_metric_key_is_canonical(self, mod):
assert mod.METRIC_KEY == "VaR_breach_rate_99_oos_regime_cond"
# ---------------------------------------------------------------------------
# kupiec_pvalue
# ---------------------------------------------------------------------------
class TestKupiecPValue:
def test_perfectly_calibrated(self, mod):
# p_hat == p0 → LR=0 → p-value=1
p = mod.kupiec_pvalue(100, 1, p0=0.01)
assert p > 0.99, f"p={p}"
def test_strong_reject_high_breach(self, mod):
# 20% breach when 1% expected → p << 0.05
p = mod.kupiec_pvalue(100, 20, p0=0.01)
assert p < 0.001, f"p={p}"
def test_zero_breaches_not_nan(self, mod):
p = mod.kupiec_pvalue(100, 0, p0=0.01)
assert not math.isnan(p)
assert 0 <= p <= 1.0
def test_all_breaches_not_nan(self, mod):
p = mod.kupiec_pvalue(10, 10, p0=0.01)
assert not math.isnan(p)
assert p < 0.001 # extremely unlikely
def test_zero_observations(self, mod):
p = mod.kupiec_pvalue(0, 0)
assert p == 1.0
+46 -8
View File
@@ -35,6 +35,8 @@ PHASE1_LR = float(_os.environ.get("JEPA_PHASE1_LR", 1e-3))
PHASE1_JOINT = bool(int(_os.environ.get("JEPA_PHASE1_JOINT", 1)))
PHASE1_JOINT_EPOCHS= int(_os.environ.get("JEPA_PHASE1_JOINT_EPOCHS", 30))
PHASE1_ENCODER_LR = float(_os.environ.get("JEPA_PHASE1_ENCODER_LR", 3e-6))
USE_MULTIPAIR = bool(int(_os.environ.get("JEPA_USE_MULTIPAIR", 0)))
JEPA_ENABLE_REGIME = bool(int(_os.environ.get("JEPA_ENABLE_REGIME", 0)))
SEED = int(_os.environ.get("JEPA_SEED", 0))
# ---------------------------
@@ -149,19 +151,45 @@ def build():
falls back to eurusd_daily.parquet otherwise.
"""
import os
hourly_path = "data/processed/eurusd_hourly.parquet"
daily_path = "data/processed/eurusd_daily.parquet"
if USE_HOURLY and os.path.exists(hourly_path):
multipair_path = "data/processed/eurusd_multipair.parquet"
hourly_path = "data/processed/eurusd_hourly.parquet"
daily_path = "data/processed/eurusd_daily.parquet"
if USE_MULTIPAIR and os.path.exists(multipair_path):
df = pd.read_parquet(multipair_path).reset_index(drop=True)
df["date"] = pd.to_datetime(df["datetime"])
# All {pair}_ret + {pair}_rv columns as features; eurusd_rv as target
feat_cols = [c for c in df.columns if c.endswith("_ret") or c.endswith("_rv")]
FEAT_COLS = feat_cols
target_col = "eurusd_rv"
elif USE_HOURLY and os.path.exists(hourly_path):
df = pd.read_parquet(hourly_path).reset_index(drop=True)
df["date"] = pd.to_datetime(df["datetime"])
# 2-channel default (HPO: adding hl_range+ret_intrabar hurt — correlated with base feats)
FEAT_COLS = ["ret", "realized_vol"]
target_col = "realized_vol"
else:
df = pd.read_parquet(daily_path).reset_index(drop=True)
df["date"] = pd.to_datetime(df["date"])
# 2-channel default (HPO: adding hl_range+ret_intrabar hurt — correlated with base feats)
# To experiment: change to ["ret", "realized_vol", "hl_range", "ret_intrabar"]
FEAT_COLS = ["ret", "realized_vol"]
FEAT_COLS = ["ret", "realized_vol"]
target_col = "realized_vol"
# ── REGIME CONDITIONING SEAM — agent may vary this mechanism ─────────────
# Baseline: concat regime flag as an additional feature channel (0=calm, 2=crisis).
# Agent may swap for FiLM conditioning, learned regime embedding, or gating.
_regime_path = "data/processed/eurusd_regime.parquet"
if JEPA_ENABLE_REGIME and os.path.exists(_regime_path):
_rdf = pd.read_parquet(_regime_path)
_ts_col = "datetime" if "datetime" in _rdf.columns else "date"
_rdf[_ts_col] = pd.to_datetime(_rdf[_ts_col])
df = df.copy()
df = df.merge(
_rdf.rename(columns={_ts_col: "date"})[["date", "regime"]],
on="date", how="left",
)
df["regime"] = df["regime"].fillna(0).astype(np.float32)
FEAT_COLS = list(FEAT_COLS) + ["regime"]
# ── END REGIME SEAM ───────────────────────────────────────────────────────
feats = df[FEAT_COLS].to_numpy(np.float32)
target = df["realized_vol"].to_numpy(np.float32)
target = df[target_col].to_numpy(np.float32)
tr_idx = df.index[df["date"].dt.year <= 2021].tolist()
te_idx = df.index[df["date"].dt.year >= 2022].tolist()
mu = feats[:tr_idx[-1]+1].mean(0)
@@ -287,12 +315,22 @@ def main():
phase1_r2 = float(1 - ((yte - pred_h) ** 2).sum() / ss_tot)
print("phase1_r2 = %.4f (n_test=%d)" % (phase1_r2, len(yte)))
# ── VaR EVAL BLOCK — do NOT edit (agent boundary) ───────────────────────
import sys as _sys
_sys.path.insert(0, _os.path.dirname(_os.path.abspath(__file__)))
from scripts.var_breach import var_breach_rate as _var_breach_rate, METRIC_KEY as _VAR_KEY
_var_rate, _kupiec_p = _var_breach_rate(pred_np.tolist(), yte.tolist())
print("%s=%.4f Kupiec_p=%.4f" % (_VAR_KEY, _var_rate, _kupiec_p))
# ── END VaR EVAL BLOCK ───────────────────────────────────────────────────
_metrics_out = _os.environ.get("METRICS_OUT", "metrics.json")
json.dump({
"val_vol_r2": val_vol_r2, "phase1_r2": phase1_r2, "n_test": len(yte),
_VAR_KEY: _var_rate, "kupiec_p": _kupiec_p,
"knobs": {"WINDOW": WINDOW, "PATCH_LEN": PATCH_LEN,
"D_MODEL": D_MODEL, "DEPTH": DEPTH, "ALPHA": ALPHA,
"DELTA_T_MAX": DELTA_T_MAX, "EPOCHS": EPOCHS},
}, open("metrics.json", "w"), indent=2)
}, open(_metrics_out, "w"), indent=2)
print("val_vol_r2 = %.4f (n_test=%d, dev=%s)" % (val_vol_r2, len(yte), dev))
# ── EXPORT BLOCK — do NOT edit (agent boundary) ──────────────────────────