# DamBackTest

### Keep coordination small. Run the data work where it belongs.

A Go distributed-systems prototype for chunk-based market-data backtesting. DamBackTest separates a **Raft-replicated metadata control plane**, **RPC execution workers**, and **checksum-verified data storage**.

[Quickstart](#quickstart) · [Architecture](#architecture) · [Capabilities](#where-it-shines) · [HTTP API](#http-api) · [Tests](#tests-and-performance) · [Limits](#current-boundaries)

**Status:** CMU coursework / research prototype. The bundled strategy is an execution example, not a realistic trading simulator. This checkout uses **Go 1.25.6** and a multi-module `go.work` workspace.

## What it does

1. Store timestamped market events in a DataNode as a named chunk.
2. Register the chunk's instrument, time range, and replica locations with the coordinator leader.
3. Submit a job for an instrument, time interval, and strategy.
4. Split matching chunks across worker groups. Workers fetch data directly from DataNodes, filter events, sort by timestamp, and execute the strategy.
5. Merge partial outputs and query job status and results through HTTP.

The result contains a job ID, event count, example PnL, and logs. Raw market events stay off the coordinator's scheduling path.

## Architecture

```mermaid
flowchart TB
    U[Client / HTTP JSON] -->|register chunk, submit job| L
    subgraph CONTROL[Metadata control plane]
      L[Coordinator leader] <-->|Raft log replication| F1[Coordinator follower]
      L <-->|Raft log replication| F2[Coordinator follower]
    end
    L -->|chunk group / RPC| W1[Worker 1]
    L -->|chunk group / RPC| W2[Worker 2]
    W1 -->|GetChunk / RPC| D1[DataNode A]
    W2 -->|GetChunk / RPC| D1
    W2 -.->|fallback location| D2[DataNode B, if populated]
    D1 --- S1[(CSV chunks + manifest + SHA-256)]
    D2 --- S2[(Replica files + manifest)]
    W1 -->|partial result| M[Merge results]
    W2 -->|partial result| M
    M --> L
    L -->|status / result| U
```

The default demo has three coordinators, two workers, and one DataNode. Additional replica locations must be populated explicitly; the system does not automatically copy or repair replicas.

```mermaid
sequenceDiagram
    participant C as Client
    participant D as DataNode
    participant L as Leader + Raft peers
    participant W as Worker group
    C->>D: PutChunk(events)
    D->>D: Write CSV + checksum + manifest
    C->>L: POST /register_chunk
    L->>L: Replicate metadata command
    C->>L: POST /submit_job
    L-->>C: job_id
    L->>W: RunBacktest(job, chunk group)
    W->>D: GetChunk
    D->>D: Verify SHA-256
    D-->>W: Market events
    W->>W: Filter, sort, run example strategy
    W-->>L: Partial result
    L->>L: Merge and replicate completion
    C->>L: GET /job_result?id=...
    L-->>C: events_read, pnl, logs
```

## Where it shines

| Design choice | Why it matters | Implementation |
|---|---|---|
| Separate control and data paths | The coordinator handles job and chunk metadata; workers retrieve the event payloads | `coordinator.go`, `worker_rpc.go` |
| Parallel chunk groups | Execute separate groups concurrently instead of routing all work through one worker | `splitChunks`, `runJobParallel` |
| Worker retries | Try other configured workers when an execution attempt fails | `runJobWithRetry` |
| Replica fallback | Try registered DataNode addresses in sequence until a chunk can be read | `fetchChunkFromReplicas` |
| Integrity checks | Detect a file that no longer matches its recorded checksum before serving it | `DataNodeRPCServer.GetChunk` |
| Explicit metadata replication | Study leader election and replicated chunk/job/result state independently of storage | `raft/`, `raft_cmd.go` |

Best suited to learning distributed execution, experimenting with failure paths, and building a small, inspectable research infrastructure prototype. These are architectural strengths, not a claim of production readiness or measured superiority over other engines.

## Quickstart

### 1. Get the code and verify the toolchain

```sh
git clone https://github.com/Yunfan-Wang/DamBackTest.git
cd DamBackTest
go version
```

Use Go **1.25.6 or a compatible newer toolchain**. Keep the checked-in `go.work`; there is no root `go.mod`. Run commands below from the repository root. The existing integration-test harness uses Windows-specific process flags and `taskkill`; use Windows for that harness as currently written.

### 2. Start the local cluster

Open six terminals at the repository root. Run one command per terminal:

```sh
# Terminal 1: chunk storage
go run ./cmd/datanode_rpc --addr 127.0.0.1:9401 --data ./data/quickstart

# Terminal 2: execution worker
go run ./cmd/worker_rpc --addr 127.0.0.1:9301

# Terminal 3: execution worker
go run ./cmd/worker_rpc --addr 127.0.0.1:9302

# Terminal 4: coordinator 0
go run ./cmd/coordinator --id 0 --addr 127.0.0.1:9000 --workers 127.0.0.1:9301,127.0.0.1:9302

# Terminal 5: coordinator 1
go run ./cmd/coordinator --id 1 --addr 127.0.0.1:9001 --workers 127.0.0.1:9301,127.0.0.1:9302

# Terminal 6: coordinator 2
go run ./cmd/coordinator --id 2 --addr 127.0.0.1:9002 --workers 127.0.0.1:9301,127.0.0.1:9302
```

Coordinators default to Raft ports **8001–8003** and control ports **8101–8103**. Leave all six terminals running and allow a leader to be elected. All addresses are loopback for local use.

### 3. Upload the sample chunk

In a seventh terminal:

```sh
go run ./cmd/put_chunk_rpc --addr 127.0.0.1:9401 --file tests/manual/basic/chunk.json
```

This sample contains three AAPL prices: **100 → 101 → 103**, at timestamps 1, 2, and 3. Upload uses the custom RPC client, not an HTTP request to the RPC port.

### 4. Register, submit, and inspect (PowerShell)

The supplied legacy `register.json` points at port 9101; use this payload for the RPC demo on **9401**. Registration is tried against each coordinator because a follower rejects writes.

```powershell
$registration = '{"chunk":{"chunk_id":"chunk1","instrument":"AAPL","start_ts":1,"end_ts":3,"replicas":["127.0.0.1:9401"],"version":1,"sealed":true}}'
$leader = $null
foreach ($port in 9000, 9001, 9002) {
    try {
        $base = "http://127.0.0.1:$port"
        Invoke-RestMethod "$base/register_chunk" -Method Post -ContentType 'application/json' -Body $registration -ErrorAction Stop | Out-Null
        $leader = $base
        break
    } catch { Write-Host "Coordinator $port did not accept registration." }
}
if (-not $leader) { throw 'No leader accepted registration. Check all coordinator terminals, wait for election, then retry.' }
$job = Invoke-RestMethod "$leader/submit_job" -Method Post -ContentType 'application/json' -Body '{"instrument":"AAPL","start_ts":1,"end_ts":3,"strategy":"momentum"}'
$status = $null
for ($attempt = 0; $attempt -lt 60; $attempt++) {
    $status = Invoke-RestMethod "$leader/job_status?id=$($job.job_id)"
    if ($status.status -in 'done', 'failed') { break }
    Start-Sleep -Milliseconds 500
}
if ($status.status -ne 'done') { throw "Job did not finish successfully: $($status.status)" }
Invoke-RestMethod "$leader/job_result?id=$($job.job_id)"
```

Expected fields: **`events_read: 3`**, **`pnl: 3`**. The generated job ID varies. This example sums positive adjacent price differences; it does not represent a tradable strategy's investment return. One chunk is assigned to one worker group, even when two workers are available.

Stop the six processes with Ctrl+C when finished. The DataNode files remain in `data/quickstart`; coordinator metadata is not durable across a complete cluster restart. Re-register chunks after restarting.

## HTTP API

| Method | Endpoint | Purpose |
|---|---|---|
| POST | `/register_chunk` | Register a `{"chunk": {...}}` payload on the leader |
| POST | `/submit_job` | Submit instrument, inclusive start/end timestamps, and strategy; receive a job ID |
| GET | `/job_status?id=<job_id>` | Inspect `queued`, `running`, `done`, or `failed` |
| GET | `/job_result?id=<job_id>` | Retrieve the merged result |

DataNode `PutChunk` / `GetChunk` and worker `RunBacktest` are **custom RPC** interfaces. Legacy HTTP worker/DataNode entry points also exist; do not mix their ports with the RPC setup above. If leadership changes during the example, discover the new leader again before submitting further writes.

## Tests and performance

```sh
# Windows integration and failure scenarios
go test -C dambt -v -timeout 180s -run TestFinalDAMBT -count=1

# Small end-to-end timing fixture
go test -C dambt -v -timeout 180s -run TestPerfDAMBT -count=1

# Build an executable without relying on a nonexistent root module
go build -o coordinator.exe ./cmd/coordinator
```

The suites cover functional behavior, failure handling, Raft metadata consistency, and performance smoke checks. The performance fixture submits **20 jobs × 4 events**, then measures throughput, completion latency, and status-query overhead. It is a tiny control-path fixture, not a large-market-data benchmark. Go race checking also needs a supported C toolchain (`-race`).

When publishing results, include the commit and local modifications, Go version, OS, CPU, dataset size, worker count, test command, and repeated-run distribution. Do not treat the Makefile's printed proposed point totals as an independently verified performance result.

### Local verification snapshot

On 22 September 2026, all five suite entry points passed on Windows amd64 / Go 1.25.6. Checkout: `e8240e3`, with a pre-existing local modification to `dambt/dambt_perf_test.go`. One run, not a statistical benchmark:

| Measurement | Observed |
|---|---|
| Timing fixture | 20 jobs / 80 event reads |
| Fixture elapsed time | 2.223 s |
| Average completion latency | 1.635 s |
| Maximum completion latency | 2.223 s |
| Average status query, 100 queries | 145.274 µs |

The harness currently overwrites its supplied worker list with generated live-worker addresses. Consequently, the named dead-worker test passing does not independently prove the intended dead-worker injection happened. Worker retry logic is present in the implementation; strengthen this test before using it as fault-injection evidence.

## Current boundaries

- **In-memory consensus state:** Raft does not restore durable state. Coordinator WAL/checkpoint code is commented out. A full cluster restart is not durable recovery.
- **Partition boundaries matter:** `MergeResults` sums partial results. It does not carry strategy state or adjacent prices between groups, so multi-worker PnL can differ from a single globally ordered run.
- **Illustrative strategy only:** `momentum` sums upward adjacent price changes. No positions, execution costs, slippage, order book, or risk model; unknown strategy names currently produce zero PnL.
- **Replica locations, not managed replication:** Fallback reads exist, but automatic replica placement and repair do not. Checksums detect corruption; they do not repair it.
- **Bounded scalability:** Workers load chunk events in memory; groups are balanced by chunk count, not bytes or locality. The coordinator scheduler processes queued jobs through a single loop, with parallelism inside each job.
- **Ordering:** Events sort by timestamp; equal timestamps have no explicit stable tie-break key.
- **Local trusted environment:** No authentication/TLS layer is provided for the demo interfaces.
- **Implemented in Go:** The reviewed active path does not contain the C++ engine or plugin architecture described in broader design goals.

## Repository map

```text
cmd/                 Coordinator, RPC worker, DataNode, upload client
dambt/coordinator.go HTTP API, scheduling, partitioning, worker retries
dambt/worker*.go     Chunk retrieval, replica fallback, filtering
dambt/datanode*.go   CSV storage, persistent manifest, checksums
dambt/strategy.go   Example strategy and partial-result merge
raft/               Consensus implementation
remote/             Custom RPC transport
tests/manual/       Example requests (some use legacy HTTP ports)
designManuals/      Design history and implementation notes
dambt-doc.md         Generated API reference
```

## Extending the system

Useful next steps are durable Raft storage, boundary-aware strategy execution, a stable event tie-break key, replica repair, and streaming/cached chunk reads. For a new strategy, start in `RunStrategy` and define whether its state can be partitioned before using `MergeResults`.

Contributions should include a reproducible scenario and a focused correctness test. No license grant is inferred from repository visibility; add an explicit license before representing this as a licensed open-source package.