diff --git a/.changeset/mcp-server.md b/.changeset/mcp-server.md new file mode 100644 index 0000000..6c2ab09 --- /dev/null +++ b/.changeset/mcp-server.md @@ -0,0 +1,45 @@ +--- +"ohlc-resample": minor +--- + +Add a **zero-dependency MCP server** (`ohlc-resample-mcp` bin) and fix a +streaming CSV→CSV output bug. + +**MCP server (bundled, no second package)** + +- New `ohlc-resample-mcp` binary ships inside this same package — install + `ohlc-resample` and you get the MCP server for free. No separate publish. +- It is a **thin adapter over the package's own CLI** (`runCli` in-process + with injected streams): it has no resampling/parsing/formatting logic of its + own, so every new CLI feature is inherited with zero maintenance. A tool + call just builds an argv array and runs the CLI, returning stdout or a file + path. +- **No `@modelcontextprotocol/sdk` dependency** — MCP is plain JSON-RPC 2.0 + over stdio, so the server hand-rolls the tiny protocol (initialize, + `tools/list`, `tools/call`, `ping`) directly. Runtime deps stay just `mri` + and `hyparquet`. +- Two tools: `resample_ohlcv_file` (takes an `input_path` + (csv/json/jsonl/ndjson/parquet), `base_timeframe`, `new_timeframe`, + `format`, `shape`, optional `map` (e.g. CCXT `timestamp`/`amount`), and + optional `output_path`) and `audit_ohlcv_file` (takes an `input_path` and + optional `map`, returns a trust report). Both are file-path-based, so the + LLM pays tokens for intent, not payload. + +**New `audit` capability** + +- New library function `auditOhlcv` + `AuditReport`: validates and describes + market-data input in one streaming pass, answering "can I trust the output + of this resampling, and exactly why?". Reports record count, time range, + source timeframe (modal positive interval), ordering (sorted, out-of-order + count, max lateness), duplicate timestamps, OHLC integrity violations, bad + values (NaN/Infinity/negative prices/volume), and missing bars. +- Wired to the CLI as `--audit` (prints a JSON report instead of resampling) + and exposed to MCP as `audit_ohlcv_file`, which inherits the CLI path with + zero extra logic. + +**Bug fix** + +- Streaming CSV → CSV output previously failed with `shaped.join is not a + function`: `IncrementalWriter` rendered object-shaped candles as CSV rows + without converting to the 6-tuple shape. CSV rows are now always written in + array shape regardless of the requested JSON shape. diff --git a/README.md b/README.md index 06fdefd..bc3241d 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,6 @@

ohlc-resample 🕯️

-Turn trade, tick, or OHLCV data into clean candlestick charts on any time frame +Transform, resample, and stream market data at any scale

@@ -298,6 +298,7 @@ Options: -b, --base-timeframe Base timeframe in seconds (default: "60") -n, --new-timeframe New timeframe in seconds (default: "300") --map Feed data as-is; map fields to canonical keys (e.g. time=timestamp,volume=amount) + --audit Audit the input instead of resampling (print a JSON trust report) -h, --help Display help for command ``` @@ -359,6 +360,61 @@ cat data.json | ohlc cat data.csv | ohlc --input-format csv ``` +## Transform and stream market data, straight from your AI + +Tell your AI assistant to reshape your market data and it just does it. Point +it at your feed, name the time frame you want, and it hands back clean output +at any scale. It already speaks the street: CCXT-style feeds, tick data, +Parquet exports, gaps in the series, files too big for memory. You ask, it +delivers. + +Set it up in one shot, then it works with whichever assistant you use: + +```bash +npm i -g ohlc-resample +``` + +That installs both the `ohlc` command and the MCP server. Most assistants +auto-detect it; for the ones that need a nudge, add a server entry: + +```json +{ + "mcpServers": { + "ohlc-resample": { + "command": "ohlc-resample-mcp", + "args": [] + } + } +} +``` + +Your assistant gets two tools. + +**`resample_ohlcv_file`** reshapes your data. Give it the file to read and +the time frame you want, and it does the rest: + +- 1-minute bars into 5-minute (or any coarser frame) +- raw trades or ticks into clean OHLCV +- CCXT `timestamp` / `amount` data without renaming a thing +- CSV, JSON, JSONL, or Parquet in; JSON, CSV, or JSONL out + +Pass an output file and it writes there, or let it return the data directly. +No need to paste anything into the chat. + +Example: resample 1-minute bars to 5-minute and save them. + +`resample_ohlcv_file(input_path: "data.csv", base_timeframe: 60, new_timeframe: 300, output_path: "out.json")` + +**`audit_ohlcv_file`** tells you whether you can trust the source before you +resample it, and exactly why. Point it at the same file and it reports the +record count, the time span, the source time frame, any out-of-order or +duplicate bars, bars whose high/low/open/close don't add up, NaN or negative +values, and bars that are simply missing. + +Example: check a feed before committing to a resample. + +`audit_ohlcv_file(input_path: "data.csv")` + ## Contributors 👤 **Adil Shaikh (https://adils.me)** diff --git a/__tests__/audit.ts b/__tests__/audit.ts new file mode 100644 index 0000000..4ea12aa --- /dev/null +++ b/__tests__/audit.ts @@ -0,0 +1,202 @@ +"use strict"; + +import { test, describe, expect, beforeAll, afterEach, afterAll } from "vitest"; +import * as fs from "fs"; +import * as path from "path"; +import * as os from "os"; +import { Writable } from "stream"; +import { auditOhlcv } from "../src"; +import { runCli } from "../src/cli"; +import { dispatch } from "../src/mcp"; +import { withTimeout } from "./utils"; +import type { IOHLCV } from "../src/types"; + +const clean: IOHLCV[] = [ + { time: 1609459200000, open: 100, high: 105, low: 95, close: 102, volume: 1000 }, + { time: 1609459260000, open: 102, high: 107, low: 101, close: 106, volume: 1200 }, + { time: 1609459320000, open: 106, high: 108, low: 104, close: 105, volume: 800 }, + { time: 1609459380000, open: 105, high: 106, low: 103, close: 104, volume: 900 }, + { time: 1609459440000, open: 104, high: 105, low: 102, close: 103, volume: 1100 }, +]; + +async function* iter(data: IOHLCV[]) { + for (const c of data) yield c; +} + +describe("auditOhlcv", () => { + test("reports a clean 1-minute series", async () => { + const r = await auditOhlcv(iter(clean)); + expect(r.records).toBe(5); + expect(r.timeRange.startMs).toBe(1609459200000); + expect(r.timeRange.endMs).toBe(1609459440000); + expect(r.timeRange.spanMs).toBe(240000); + expect(r.baseTimeframe).toBe(60); + expect(r.ordering).toEqual({ sorted: true, outOfOrder: 0, maxLatenessMs: 0 }); + expect(r.duplicates.duplicateTimestamps).toBe(0); + expect(r.ohlc.invalidBars).toBe(0); + expect(r.values).toEqual({ nan: 0, infinity: 0, negativePrices: 0, negativeVolume: 0 }); + expect(r.gaps).toEqual({ expectedBars: 5, observed: 5, missing: 0 }); + }); + + test("flags out-of-order, duplicate, invalid-bar, and bad-value records", async () => { + const messy: IOHLCV[] = [ + { time: 1609459200000, open: 1, high: 2, low: 0, close: 1.5, volume: 10 }, + { time: 1609459260000, open: 1, high: 2, low: 3, close: 1, volume: 5 }, // high < low + { time: 1609459200000, open: 1, high: 2, low: 0, close: 1.5, volume: 10 }, // duplicate + { time: 1609459380000, open: 2, high: 3, low: 1, close: 2, volume: -1 }, // negative volume + { time: 1609459320000, open: 1, high: 2, low: 0, close: 1, volume: 8 }, // out-of-order + { time: 1609459440000, open: NaN, high: 2, low: 0, close: 1, volume: 8 }, // NaN + ]; + const r = await auditOhlcv(iter(messy)); + expect(r.ordering.sorted).toBe(false); + expect(r.ordering.outOfOrder).toBeGreaterThan(0); + expect(r.ordering.maxLatenessMs).toBeGreaterThan(0); + expect(r.duplicates.duplicateTimestamps).toBe(1); + expect(r.ohlc.invalidBars).toBe(1); + expect(r.values.negativeVolume).toBe(1); + expect(r.values.nan).toBeGreaterThan(0); + }); + + test("computes missing bars from the modal timeframe", async () => { + const gappy: IOHLCV[] = [ + { time: 1609459200000, open: 1, high: 2, low: 0, close: 1, volume: 10 }, + { time: 1609459260000, open: 1, high: 2, low: 0, close: 1, volume: 10 }, + { time: 1609459380000, open: 1, high: 2, low: 0, close: 1, volume: 10 }, + { time: 1609459440000, open: 1, high: 2, low: 0, close: 1, volume: 10 }, + ]; + const r = await auditOhlcv(iter(gappy)); + expect(r.baseTimeframe).toBe(60); + expect(r.gaps.expectedBars).toBe(5); + expect(r.gaps.observed).toBe(4); + expect(r.gaps.missing).toBe(1); + }); + + test("honors an explicit baseTimeframe option", async () => { + const r = await auditOhlcv(iter(clean), { baseTimeframe: 300 }); + expect(r.baseTimeframe).toBe(300); + expect(r.gaps.expectedBars).toBe(1); + }); + + test("applies a field map to object records", async () => { + const mapped = clean.slice(0, 2).map((c) => ({ + timestamp: c.time, + amount: c.volume, + open: c.open, + high: c.high, + low: c.low, + close: c.close, + })); + const r = await auditOhlcv(iter(mapped as unknown as IOHLCV[]), { + map: { time: "timestamp", volume: "amount" }, + }); + expect(r.records).toBe(2); + expect(r.timeRange.startMs).toBe(1609459200000); + expect(r.values.negativeVolume).toBe(0); + }); + + test("audits a parquet file path directly", async () => { + const p = path.join(import.meta.dirname, "fixtures", "ohlcv.parquet"); + const r = await auditOhlcv(p); + expect(r.records).toBeGreaterThan(0); + expect(r.baseTimeframe).toBeGreaterThan(0); + expect(r.ohlc.invalidBars).toBe(0); + }); + + test("throws on empty input", async () => { + await expect(auditOhlcv(iter([]))).rejects.toThrow("no candles"); + }); +}); + +describe("CLI --audit and MCP audit_ohlcv_file", () => { + let tempDir: string; + let csvPath: string; + let messyJsonlPath: string; + + function captureWritable() { + let data = ""; + const writable = new Writable({ + write(chunk, _encoding, callback) { + data += chunk.toString(); + callback(); + }, + }); + return { writable, getData: () => data }; + } + + beforeAll(() => { + tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "audit-test-")); + const csvRows = clean.map((c) => `${c.time},${c.open},${c.high},${c.low},${c.close},${c.volume}`).join("\n"); + csvPath = path.join(tempDir, "ohlcv.csv"); + fs.writeFileSync(csvPath, "time,open,high,low,close,volume\n" + csvRows); + + const messy = [ + { time: 1609459200000, open: 1, high: 2, low: 0, close: 1.5, volume: 10 }, + { time: 1609459260000, open: 1, high: 2, low: 3, close: 1, volume: 5 }, + { time: 1609459200000, open: 1, high: 2, low: 0, close: 1.5, volume: 10 }, + { time: 1609459380000, open: 2, high: 3, low: 1, close: 2, volume: -1 }, + { time: 1609459320000, open: 1, high: 2, low: 0, close: 1, volume: 8 }, + ]; + messyJsonlPath = path.join(tempDir, "messy.jsonl"); + fs.writeFileSync(messyJsonlPath, messy.map((c) => JSON.stringify(c)).join("\n")); + }); + + afterEach(async () => { + process.exitCode = 0; + }); + + test("prints a JSON trust report for a CSV file", async () => { + await withTimeout(async () => { + const out = captureWritable(); + const err = captureWritable(); + await runCli(["node", "cli.js", "-i", csvPath, "--audit"], undefined, out.writable, err.writable); + const r = JSON.parse(out.getData()); + expect(r.format).toBe("csv"); + expect(r.records).toBe(5); + expect(r.baseTimeframe).toBe(60); + expect(r.schema).toEqual(["time", "open", "high", "low", "close", "volume"]); + expect(r.gaps.missing).toBe(0); + }, 1000, "audit csv"); + }); + + test("flags issues in a messy JSONL file", async () => { + await withTimeout(async () => { + const out = captureWritable(); + const err = captureWritable(); + await runCli(["node", "cli.js", "-i", messyJsonlPath, "--audit"], undefined, out.writable, err.writable); + const r = JSON.parse(out.getData()); + expect(r.format).toBe("jsonl"); + expect(r.ordering.sorted).toBe(false); + expect(r.ordering.outOfOrder).toBeGreaterThan(0); + expect(r.duplicates.duplicateTimestamps).toBeGreaterThan(0); + expect(r.ohlc.invalidBars).toBeGreaterThan(0); + expect(r.values.negativeVolume).toBeGreaterThan(0); + }, 1000, "audit messy jsonl"); + }); + + test("audits a parquet file", async () => { + await withTimeout(async () => { + const p = path.join(import.meta.dirname, "fixtures", "ohlcv.parquet"); + const out = captureWritable(); + const err = captureWritable(); + await runCli(["node", "cli.js", "-i", p, "--audit"], undefined, out.writable, err.writable); + const r = JSON.parse(out.getData()); + expect(r.format).toBe("parquet"); + expect(r.records).toBeGreaterThan(0); + }, 1000, "audit parquet"); + }); + + test("MCP audit_ohlcv_file returns an audit report", async () => { + const res = await dispatch({ + jsonrpc: "2.0", + id: 1, + method: "tools/call", + params: { name: "audit_ohlcv_file", arguments: { input_path: csvPath } }, + }) as any; + const r = JSON.parse(res.content[0].text); + expect(r.records).toBe(5); + expect(r.baseTimeframe).toBe(60); + expect(r.timeRange.startMs).toBe(1609459200000); + }); + + afterAll(() => fs.rmSync(tempDir, { recursive: true, force: true })); +}); diff --git a/__tests__/cli.ts b/__tests__/cli.ts index 37a52e8..f797b1b 100644 --- a/__tests__/cli.ts +++ b/__tests__/cli.ts @@ -333,6 +333,20 @@ describe('CLI', () => { }, 1000, 'stream CSV to JSONL'); }); + test('streams CSV file -> CSV output (object shape to CSV rows)', async () => { + await withTimeout(async () => { + const out = captureWritable(); + const err = captureWritable(); + await runCli(['node', 'cli.js', '-i', csvPath, '-f', 'csv'], undefined, out.writable, err.writable); + const lines = out.getData().trim().split('\n'); + expect(lines[0]).toBe('time,open,high,low,close,volume'); + expect(lines[1]).toBe( + `${expectedCandle.time},${expectedCandle.open},${expectedCandle.high},${expectedCandle.low},${expectedCandle.close},${expectedCandle.volume}` + ); + expect(process.exitCode).toBe(0); + }, 1000, 'stream CSV to CSV'); + }); + test('streams JSONL file input', async () => { await withTimeout(async () => { const jsonlPath = path.join(tempDir, 'test.jsonl'); diff --git a/__tests__/mcp.ts b/__tests__/mcp.ts new file mode 100644 index 0000000..64dffe6 --- /dev/null +++ b/__tests__/mcp.ts @@ -0,0 +1,147 @@ +import { describe, test, expect, beforeEach, afterEach } from 'vitest'; +import * as fs from 'fs'; +import * as path from 'path'; +import * as os from 'os'; +import { IOHLCV } from '../src/types'; +import { withTimeout } from './utils'; +import { dispatch } from '../src/mcp'; + +describe('MCP server (stdio JSON-RPC, CLI-delegated)', () => { + let tempDir: string; + let jsonPath: string; + let csvPath: string; + + const testData: IOHLCV[] = [ + { time: 1609459200000, open: 100, high: 105, low: 95, close: 102, volume: 1000 }, + { time: 1609459260000, open: 102, high: 107, low: 101, close: 106, volume: 1200 }, + { time: 1609459320000, open: 106, high: 108, low: 104, close: 105, volume: 800 }, + { time: 1609459380000, open: 105, high: 106, low: 103, close: 104, volume: 900 }, + { time: 1609459440000, open: 104, high: 105, low: 102, close: 103, volume: 1100 }, + ]; + + beforeEach(async () => { + tempDir = await fs.promises.mkdtemp(path.join(os.tmpdir(), 'ohlc-mcp-test-')); + jsonPath = path.join(tempDir, 'in.json'); + csvPath = path.join(tempDir, 'in.csv'); + await fs.promises.writeFile(jsonPath, JSON.stringify(testData, null, 2)); + const csvContent = 'time,open,high,low,close,volume\n' + + testData.map(d => `${d.time},${d.open},${d.high},${d.low},${d.close},${d.volume}`).join('\n'); + await fs.promises.writeFile(csvPath, csvContent); + }); + + afterEach(async () => { + await fs.promises.rm(tempDir, { recursive: true, force: true }); + }); + + test('initialize advertises capabilities and version', async () => { + const res = await dispatch({ + jsonrpc: '2.0', + id: 1, + method: 'initialize', + params: { protocolVersion: '2024-11-05' }, + }) as any; + expect(res.serverInfo.name).toBe('ohlc-resample-mcp'); + expect(typeof res.serverInfo.version).toBe('string'); + expect(res.capabilities.tools).toBeTruthy(); + }); + + test('tools/list exposes the resample and audit tools with schemas', async () => { + const res = await dispatch({ jsonrpc: '2.0', id: 2, method: 'tools/list', params: {} }) as any; + expect(res.tools).toHaveLength(2); + const names = res.tools.map((t: { name: string }) => t.name).sort(); + expect(names).toEqual(['audit_ohlcv_file', 'resample_ohlcv_file']); + const resample = res.tools.find((t: { name: string }) => t.name === 'resample_ohlcv_file'); + expect(resample.inputSchema.required).toContain('input_path'); + expect(resample.inputSchema.properties.new_timeframe.default).toBe(300); + const audit = res.tools.find((t: { name: string }) => t.name === 'audit_ohlcv_file'); + expect(audit.inputSchema.required).toContain('input_path'); + expect(audit.inputSchema.properties.map).toBeTruthy(); + }); + + test('resamples a JSON file and returns candles as text', async () => { + const res = await dispatch({ + jsonrpc: '2.0', + id: 3, + method: 'tools/call', + params: { + name: 'resample_ohlcv_file', + arguments: { input_path: jsonPath, base_timeframe: 60, new_timeframe: 300, shape: 'object' }, + }, + }) as any; + expect(res.isError).toBeFalsy(); + const output = JSON.parse(res.content[0].text); + expect(output).toHaveLength(1); + expect(output[0]).toMatchObject({ + time: 1609459200000, open: 100, high: 108, low: 95, close: 103, volume: 5000, + }); + }); + + test('streams CSV input -> CSV output (object shape to CSV rows)', async () => { + const res = await dispatch({ + jsonrpc: '2.0', + id: 4, + method: 'tools/call', + params: { + name: 'resample_ohlcv_file', + arguments: { input_path: csvPath, base_timeframe: 60, new_timeframe: 300, format: 'csv' }, + }, + }) as any; + expect(res.isError).toBeFalsy(); + const lines = res.content[0].text.trim().split('\n'); + expect(lines[0]).toBe('time,open,high,low,close,volume'); + expect(lines[1]).toBe('1609459200000,100,108,95,103,5000'); + }); + + test('writes to output_path and returns the path', async () => { + const outPath = path.join(tempDir, 'out.json'); + const res = await dispatch({ + jsonrpc: '2.0', + id: 5, + method: 'tools/call', + params: { + name: 'resample_ohlcv_file', + arguments: { input_path: jsonPath, base_timeframe: 60, new_timeframe: 300, output_path: outPath }, + }, + }) as any; + expect(res.isError).toBeFalsy(); + expect(res.content[0].text).toContain(outPath); + const written = JSON.parse(await fs.promises.readFile(outPath, 'utf8')); + expect(written).toHaveLength(1); + }); + + test('returns isError for a missing file', async () => { + const res = await dispatch({ + jsonrpc: '2.0', + id: 6, + method: 'tools/call', + params: { + name: 'resample_ohlcv_file', + arguments: { input_path: path.join(tempDir, 'nope.csv') }, + }, + }) as any; + expect(res.isError).toBe(true); + expect(res.content[0].text).toContain('ENOENT'); + }); + + test('returns isError for a non-multiple timeframe', async () => { + const res = await dispatch({ + jsonrpc: '2.0', + id: 7, + method: 'tools/call', + params: { + name: 'resample_ohlcv_file', + arguments: { input_path: jsonPath, base_timeframe: 60, new_timeframe: 200 }, + }, + }) as any; + expect(res.isError).toBe(true); + expect(res.content[0].text).toContain('integer multiple'); + }); + + test('rejects unknown methods with a JSON-RPC error', async () => { + await withTimeout(async () => { + await expect( + dispatch({ jsonrpc: '2.0', id: 8, method: 'tools/nope', params: {} }), + ).rejects.toMatchObject({ code: -32601 }); + }, 1000, 'unknown method'); + }); +}); diff --git a/__tests__/utils.ts b/__tests__/utils.ts index f656d2e..632ad87 100644 --- a/__tests__/utils.ts +++ b/__tests__/utils.ts @@ -65,7 +65,7 @@ describe('withTimeout utility', () => { await withTimeout( async () => { - await new Promise(resolve => setTimeout(resolve, 800)); + await new Promise(resolve => setTimeout(resolve, 900)); return 'success'; }, 1000, diff --git a/package.json b/package.json index b5bc8ef..478947c 100644 --- a/package.json +++ b/package.json @@ -22,6 +22,7 @@ "scripts": { "prebuild": "rimraf dist", "build": "tsc --build && node scripts/make-cjs-wrapper.mjs", + "mcp": "node dist/mcp.js", "build:check": "node -e \"import('./dist/index.js')\" && node -e \"const m=require('./dist/index.cjs');m.resampleOhlcv([{time:1609459200000,open:100,high:105,low:95,close:102,volume:500}],{baseTimeframe:60,newTimeframe:300})\"", "prepublishOnly": "npm test && npm run build && npm run build:check", "test": "vitest run", @@ -31,7 +32,8 @@ "release": "npm publish --no-git-checks --access public --provenance" }, "bin": { - "ohlc-resample": "./dist/cli.js" + "ohlc-resample": "./dist/cli.js", + "ohlc-resample-mcp": "./dist/mcp.js" }, "keywords": [ "ohlc", diff --git a/src/cli.ts b/src/cli.ts index 4605099..4ecd76a 100644 --- a/src/cli.ts +++ b/src/cli.ts @@ -6,7 +6,7 @@ import * as readline from 'readline'; import { createRequire } from 'module'; import { fileURLToPath } from 'node:url'; import mri from 'mri';import { IOHLCV, OHLCV } from './types.js'; -import { resampleOhlcv, resampleOhlcvAsync } from './lib.js'; +import { resampleOhlcv, resampleOhlcvAsync, auditOhlcv } from './lib.js'; import { mapToOhlcv } from './map.js'; import type { OhlcvFieldMap, OhlcvMap } from './map.js'; @@ -222,6 +222,7 @@ function prefixError(err: unknown): Error { interface ParsedInput { data: IOHLCV[] | OHLCV[]; shape: Shape; + format?: InputFormat; } function parseInput(raw: string, format: InputFormat, map?: OhlcvMap): ParsedInput { @@ -269,6 +270,7 @@ async function* csvCandleReader( lines: AsyncGenerator, onSkipped: () => void, map?: OhlcvMap, + meta?: { headers?: string[] }, ): AsyncGenerator { let headers = REQUIRED_FIELDS as readonly string[]; let headerSeen = false; @@ -286,6 +288,7 @@ async function* csvCandleReader( headers = first as readonly string[]; } started = true; + if (meta) meta.headers = headers as string[]; if (headerSeen) continue; } const candle = parseCSVLine(trimmed, headers, map); @@ -303,12 +306,22 @@ async function* jsonlCandleReader( lines: AsyncGenerator, onSkipped: () => void, map?: OhlcvMap, + meta?: { shape?: Shape; schema?: string[] }, ): AsyncGenerator { for await (const line of lines) { const trimmed = line.trim(); if (!trimmed) continue; const candle = parseJSONLLine(trimmed, map); if (candle === null) { onSkipped(); continue; } + if (meta && meta.shape === undefined) { + if (Array.isArray(candle)) { + meta.shape = 'array'; + meta.schema = [...REQUIRED_FIELDS]; + } else { + meta.shape = 'object'; + meta.schema = Object.keys(candle); + } + } yield candle; } } @@ -332,9 +345,13 @@ class IncrementalWriter { } async writeCandle(candle: OHLCV | IOHLCV): Promise { - const shaped = this.shape === 'array' + // CSV rows are always the 6-tuple shape regardless of the requested JSON + // shape, so force array shape for CSV output. + const shaped = this.format === 'csv' ? toArrayShape([candle] as OHLCV[])[0] - : toObjectShape([candle] as IOHLCV[])[0]; + : this.shape === 'array' + ? toArrayShape([candle] as OHLCV[])[0] + : toObjectShape([candle] as IOHLCV[])[0]; let text: string; if (this.format === 'csv') { text = (shaped as OHLCV).join(',') + '\n'; @@ -431,6 +448,149 @@ async function writeResampledStream( } } +interface AuditMeta { + headers?: string[]; + shape?: Shape; + schema?: string[]; +} + +/** Wrap a buffered array as an async iterable for `auditOhlcv`. */ +async function* asyncIterableFromArray(data: OHLCV[] | IOHLCV[]): AsyncGenerator { + for (const c of data) yield c; +} + +/** + * Read all of stdin into a parsed OHLCV array. Detects the format unless + * `inputFormat` is explicit. Exposes the detected format for audit reports. + */ +async function readPipeData( + stream: NodeJS.ReadableStream, + inputFormat: InputFormatOption, + map: OhlcvMap | undefined, + stderr: NodeJS.WritableStream, +): Promise { + const raw = await new Promise((resolve, reject) => { + let buf = ''; + stream.on('data', chunk => { buf += chunk; }); + stream.on('end', () => resolve(buf)); + stream.on('error', err => reject(prefixError(err))); + }); + const format: InputFormat = inputFormat === 'auto' ? detectFormat(raw) : inputFormat; + const parsed = parseInput(raw, format, map); + if (format === 'csv') { + const { skipped } = parseCSV(raw, map); + if (skipped > 0) stderr.write(`Warning: skipped ${skipped} malformed CSV row(s)\n`); + } + return { data: parsed.data, shape: parsed.shape, format }; +} + +/** Read a JSON array file into parsed OHLCV. */ +async function readJsonFileData(filePath: string, map?: OhlcvMap): Promise { + let raw: string; + try { + raw = await fs.promises.readFile(path.resolve(filePath), 'utf8'); + } catch (err) { + throw prefixError(err); + } + return parseInput(raw, 'json', map); +} + +/** + * Audit path (`--audit`): inspect the input instead of resampling it. Reuses + * the same streaming readers as the resample path, so large CSV/JSONL/Parquet + * files are audited in bounded memory. Prints a JSON trust report. + */ +async function runAudit( + input: string | undefined, + stdin: NodeJS.ReadableStream, + options: { inputFormat: InputFormatOption; map?: OhlcvMap }, + stdout: NodeJS.WritableStream, + stderr: NodeJS.WritableStream, +): Promise { + const { map } = options; + let format: InputFormat | 'parquet' = 'json'; + let shape: Shape = 'object'; + let schema: string[] = [...REQUIRED_FIELDS]; + let skipped = 0; + + // The readers below (CSV/JSONL/JSON/pipe) already apply `map` before + // yielding, so auditOhlcv must NOT re-map or canonical objects would be + // double-mapped (e.g. row['timestamp'] -> undefined -> NaN time). Only the + // Parquet string-path branch hands the raw path to auditOhlcv, which applies + // the map itself. `useMap` flags that branch. + // + // CSV/JSONL shape+schema are discovered by the reader on its first record, + // so `meta` is captured after the stream is consumed but before reporting. + const finish = async ( + iter: AsyncIterable | string, + useMap = false, + meta?: AuditMeta, + ) => { + const report = await auditOhlcv(iter, useMap ? { map } : {}); + if (meta) { + if (meta.shape) shape = meta.shape; + schema = meta.headers ?? meta.schema ?? [...REQUIRED_FIELDS]; + } + const out = { format, shape, schema, skipped, ...report }; + await writeChunk(stdout, JSON.stringify(out, null, 2)); + }; + + if (input) { + const ext = path.extname(input).slice(1).toLowerCase(); + if (!['csv', 'json', 'jsonl', 'ndjson', 'parquet'].includes(ext)) { + throw new Error('Only CSV, JSON, and Parquet files are accepted as input'); + } + const source = fs.createReadStream(path.resolve(input), { encoding: 'utf8' }); + const lines = lineReader(source); + + if (ext === 'csv') { + format = 'csv'; + const meta: AuditMeta = {}; + await finish(csvCandleReader(lines, () => skipped++, map, meta), false, meta); + return; + } + if (ext === 'jsonl' || ext === 'ndjson') { + format = 'jsonl'; + const meta: AuditMeta = {}; + await finish(jsonlCandleReader(lines, () => skipped++, map, meta), false, meta); + return; + } + if (ext === 'parquet') { + format = 'parquet'; + shape = 'object'; + schema = [...REQUIRED_FIELDS]; + await finish(input, true); + return; + } + // json array (buffered) + format = 'json'; + const parsed = await readJsonFileData(input, map); + shape = parsed.shape; + schema = + parsed.shape === 'object' && parsed.data.length > 0 + ? Object.keys(parsed.data[0]) + : ([...REQUIRED_FIELDS]); + await finish(asyncIterableFromArray(parsed.data)); + return; + } + + // Pipe input. + if (options.inputFormat === 'jsonl') { + format = 'jsonl'; + const meta: AuditMeta = {}; + await finish(jsonlCandleReader(lineReader(stdin), () => skipped++, map, meta), false, meta); + return; + } + const parsed = await readPipeData(stdin, options.inputFormat, map, stderr); + format = parsed.format ?? 'json'; + shape = parsed.shape; + schema = + parsed.shape === 'object' && parsed.data.length > 0 + ? Object.keys(parsed.data[0]) + : ([...REQUIRED_FIELDS]); + await finish(asyncIterableFromArray(parsed.data)); +} + /** * Parse the `--map` flag (Record form only; a mapping function can't be a CLI * arg) into an `OhlcvFieldMap`. Format: `field=sourceKey` entries separated by @@ -463,7 +623,7 @@ function parseMapFlag(value: string | undefined): OhlcvMap | undefined { const KNOWN_FLAGS = new Set([ 'input', 'i', 'output', 'o', 'format', 'f', 'input-format', 'shape', 's', 'base-timeframe', 'b', 'new-timeframe', 'n', - 'map', 'help', 'h', 'version', 'V', + 'map', 'audit', 'help', 'h', 'version', 'V', ]); const HELP = `Usage: ohlc-resample [options] @@ -480,6 +640,7 @@ Options: -b, --base-timeframe Base timeframe in seconds (default: "60") -n, --new-timeframe New timeframe in seconds (default: "300") --map Map record fields to canonical keys (e.g. time=timestamp,close=cl,volume=vol) + --audit audit the input instead of resampling (print a trust report) -h, --help display help for command\n`; // Tiny arg parser (mri). Unlike commander, mri is silent about unknown @@ -507,7 +668,7 @@ function parseArgs(argv: string[]) { 'input', 'output', 'format', 'input-format', 'shape', 'base-timeframe', 'new-timeframe', 'map', ], - boolean: ['help', 'version'], + boolean: ['help', 'version', 'audit'], }); const unknown = Object.keys(flags).filter(k => k !== '_' && !KNOWN_FLAGS.has(k)); if (unknown.length > 0) { @@ -550,6 +711,7 @@ export async function runCli( baseTimeframe: flags['base-timeframe'], newTimeframe: flags['new-timeframe'], map: flags.map, + audit: flags.audit, }; } catch (error: unknown) { stderr.write((error instanceof Error ? error.message : String(error)) + '\n'); @@ -557,33 +719,6 @@ export async function runCli( return; } - async function readJsonFileData(filePath: string, map?: OhlcvMap): Promise { - let raw: string; - try { - raw = await fs.promises.readFile(path.resolve(filePath), 'utf8'); - } catch (err) { - throw prefixError(err); - } - return parseInput(raw, 'json', map); - } - - async function readPipeData(stream: NodeJS.ReadableStream, map?: OhlcvMap): Promise { - const raw = await new Promise((resolve, reject) => { - let buf = ''; - stream.on('data', chunk => { buf += chunk; }); - stream.on('end', () => resolve(buf)); - stream.on('error', err => reject(prefixError(err))); - }); - const requested = options.inputFormat as InputFormatOption; - const format: InputFormat = requested === 'auto' ? detectFormat(raw) : requested; - const parsed = parseInput(raw, format, map); - if (format === 'csv') { - const { skipped } = parseCSV(raw, map); - if (skipped > 0) stderr.write(`Warning: skipped ${skipped} malformed CSV row(s)\n`); - } - return parsed; - } - async function writeOutput( data: OHLCV[] | IOHLCV[], format: OutputFormat, @@ -603,6 +738,16 @@ export async function runCli( } try { + if (options.audit) { + await runAudit( + options.input as string | undefined, + stdin, + { inputFormat: options.inputFormat as InputFormatOption, map: parseMapFlag(options.map as string | undefined) }, + stdout, + stderr, + ); + return; + } const baseTimeframe = parseInt(options.baseTimeframe, 10); const newTimeframe = parseInt(options.newTimeframe, 10); if (isNaN(baseTimeframe) || isNaN(newTimeframe)) { @@ -693,7 +838,7 @@ export async function runCli( return; } - const parsed = await readPipeData(stdin, map); + const parsed = await readPipeData(stdin, options.inputFormat as InputFormatOption, map, stderr); const outputShape: Shape = options.shape === 'auto' ? parsed.shape : options.shape; const resampled = resampleOhlcv(parsed.data as IOHLCV[], { baseTimeframe, newTimeframe }); await writeOutput(resampled, outputFormat, outputShape, options.output); diff --git a/src/lib.ts b/src/lib.ts index b1750ba..b742130 100644 --- a/src/lib.ts +++ b/src/lib.ts @@ -280,6 +280,180 @@ export async function* resampleOhlcvAsync( for (const k of keys) yield emit(buckets.get(k)!); } +/** + * Result of auditing an OHLCV input: structural checks that tell you whether + * the source is safe to resample, and exactly why (or why not). + */ +export interface AuditReport { + /** Total number of valid OHLCV records inspected. */ + records: number; + /** Min/max timestamps and the span they cover (epoch ms + ISO strings). */ + timeRange: { + startMs: number; + endMs: number; + spanMs: number; + start: string; + end: string; + }; + /** Detected source timeframe in seconds (modal interval), or `null` when unknown. */ + baseTimeframe: number | null; + ordering: { + /** True when every record arrived in ascending timestamp order. */ + sorted: boolean; + /** Count of records whose timestamp is behind the max seen so far. */ + outOfOrder: number; + /** Largest distance (ms) a late record lagged behind the max seen. */ + maxLatenessMs: number; + }; + duplicates: { + /** Count of records whose timestamp duplicates an earlier one. */ + duplicateTimestamps: number; + }; + ohlc: { + /** Count of bars that violate OHLC invariants (highmin(open,close)). */ + invalidBars: number; + }; + values: { + nan: number; + infinity: number; + negativePrices: number; + negativeVolume: number; + }; + gaps: { + /** Expected bars across the span at the base timeframe. */ + expectedBars: number; + /** Distinct timestamps observed. */ + observed: number; + /** Missing bars = expected - observed (floor 0). */ + missing: number; + }; +} + +/** + * Audit OHLCV input in a single streaming pass and return an `AuditReport`. + * Memory is O(distinct timestamps) because duplicate/gap detection needs to + * remember which bucket timestamps have been seen; everything else is O(1). + * + * @param source AsyncIterable of OHLCV tuples/IOHLCV objects, or a string + * Parquet file path (streamed row-group by row-group). + * @param options.baseTimeframe Optional known source timeframe in seconds; + * when omitted it is detected as the modal interval between records. + * @param options.map Optional per-record map (Record or function form). + * @throws on empty input. + */ +export async function auditOhlcv( + source: AsyncIterable | string, + options: { baseTimeframe?: number; map?: OhlcvMap } = {} +): Promise { + const { baseTimeframe, map } = options; + const input: AsyncIterable = + typeof source === 'string' + ? parquetOhlcvRowsAsync(source, map) + : (map ? mapOhlcvIterable(source, map) : source); + + const it = input[Symbol.asyncIterator](); + const first = await it.next(); + if (first.done) { + throw new Error('input OHLCV data has no candles'); + } + + const toTuple = (c: OHLCV | IOHLCV): OHLCV => + Array.isArray(c) + ? [Number(c[0]), Number(c[1]), Number(c[2]), Number(c[3]), Number(c[4]), Number(c[5])] + : [Number(c.time), Number(c.open), Number(c.high), Number(c.low), Number(c.close), Number(c.volume)]; + + const firstTuple = toTuple(first.value); + let records = 1; + let minTime = firstTuple[OHLCVField.TIME]; + let maxTime = firstTuple[OHLCVField.TIME]; + let maxTimeSeen = firstTuple[OHLCVField.TIME]; + let outOfOrder = 0; + let maxLatenessMs = 0; + const seen = new Set([firstTuple[OHLCVField.TIME]]); + let duplicateTimestamps = 0; + let invalidBars = 0; + let nan = 0; + let infinity = 0; + let negativePrices = 0; + let negativeVolume = 0; + const deltaHist: Record = {}; + let prev = firstTuple[OHLCVField.TIME]; + + for (let r = await it.next(); !r.done; r = await it.next()) { + const [time, open, high, low, close, volume] = toTuple(r.value); + records++; + + if (time < minTime) minTime = time; + if (time > maxTime) maxTime = time; + + if (time < maxTimeSeen) { + outOfOrder++; + const lateness = maxTimeSeen - time; + if (lateness > maxLatenessMs) maxLatenessMs = lateness; + } + maxTimeSeen = Math.max(maxTimeSeen, time); + + if (seen.has(time)) { + duplicateTimestamps++; + } else { + seen.add(time); + } + + if (time > prev) { + const sec = Math.round((time - prev) / 1000); + deltaHist[sec] = (deltaHist[sec] || 0) + 1; + } + prev = time; + + if (high < low || high < Math.max(open, close) || low > Math.min(open, close)) { + invalidBars++; + } + + if ([time, open, high, low, close, volume].some(Number.isNaN)) nan++; + if ([time, open, high, low, close, volume].some(f => f === Infinity || f === -Infinity)) infinity++; + if (open < 0 || high < 0 || low < 0 || close < 0) negativePrices++; + if (volume < 0) negativeVolume++; + } + + // Source timeframe: honor the option, else take the modal positive interval. + // Intervals that round to <=0 seconds (sub-second data) are treated as unknown. + let detected: number | null = baseTimeframe ?? null; + if (detected === null) { + let best = 0; + let bestCount = -1; + for (const [sec, count] of Object.entries(deltaHist)) { + if (count > bestCount || (count === bestCount && Number(sec) < best)) { + best = Number(sec); + bestCount = count; + } + } + detected = bestCount > 0 && best > 0 ? best : null; + } + + const spanMs = maxTime - minTime; + const expectedBars = + detected !== null && spanMs > 0 ? Math.floor(spanMs / (detected * 1000)) + 1 : 0; + const observed = seen.size; + const missing = Math.max(0, expectedBars - observed); + + return { + records, + timeRange: { + startMs: minTime, + endMs: maxTime, + spanMs, + start: new Date(minTime).toISOString(), + end: new Date(maxTime).toISOString(), + }, + baseTimeframe: detected, + ordering: { sorted: outOfOrder === 0, outOfOrder, maxLatenessMs }, + duplicates: { duplicateTimestamps }, + ohlc: { invalidBars }, + values: { nan, infinity, negativePrices, negativeVolume }, + gaps: { expectedBars, observed, missing }, + }; +} + /** * Aggregate group of ticks to one OHLCV object * @param time diff --git a/src/mcp.ts b/src/mcp.ts new file mode 100644 index 0000000..6e73bdf --- /dev/null +++ b/src/mcp.ts @@ -0,0 +1,246 @@ +#!/usr/bin/env node +/** + * ohlc-resample MCP server — zero-dependency, stdio JSON-RPC. + * + * This is a *thin adapter* over the package's own CLI (src/cli.ts). It has + * no resampling/parsing/formatting logic of its own: every tool call builds + * an argv array and runs `runCli` in-process with injected streams, exactly + * like the `ohlc` binary would. New CLI features are inherited for free. + * + * Wire protocol: MCP is JSON-RPC 2.0 over stdio, one message per line. + */ +import * as readline from 'node:readline'; +import { Writable, Readable } from 'node:stream'; +import { createRequire } from 'node:module'; +import { fileURLToPath } from 'node:url'; +import * as path from 'node:path'; +import { runCli } from './cli.js'; + +const require = createRequire(import.meta.url); +const { version: VERSION } = require('../package.json') as { version: string }; + +const SERVER_NAME = 'ohlc-resample-mcp'; +const DEFAULT_PROTOCOL_VERSION = '2024-11-05'; + +/** A writable stream that just accumulates what is written to it. */ +class CaptureStream extends Writable { + private chunks: Buffer[] = []; + _write(chunk: Buffer | string, _enc: string, cb: (err?: Error | null) => void): void { + this.chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk)); + cb(); + } + get text(): string { + return Buffer.concat(this.chunks).toString('utf8'); + } +} + +/** An empty readable stream used as the CLI's stdin when input comes from a file. */ +function emptyReadable(): Readable { + return Readable.from([]); +} + +const TOOLS = [ + { + name: 'resample_ohlcv_file', + description: + 'Resample OHLCV candle data from one time frame to a coarser one, reading from a local ' + + 'file and (optionally) writing the result to another file. Accepts CSV, JSON, JSONL, ' + + 'NDJSON, or Parquet input (CSV/JSONL/Parquet stream for large files). Supports arbitrary ' + + 'record schemas via `map` (e.g. CCXT timestamp/amount). ' + + 'Input times are epoch milliseconds. `new_timeframe` must be a positive integer multiple ' + + 'of `base_timeframe`. Returns the resampled output as text unless `output_path` is given, ' + + 'in which case it writes the file and returns the path.', + inputSchema: { + type: 'object', + properties: { + input_path: { + type: 'string', + description: 'Input file path (csv, json, jsonl, ndjson, parquet).', + }, + base_timeframe: { + type: 'number', + description: 'Source timeframe in seconds.', + default: 60, + }, + new_timeframe: { + type: 'number', + description: 'Target timeframe in seconds (integer multiple of base_timeframe).', + default: 300, + }, + format: { + type: 'string', + enum: ['json', 'csv', 'jsonl'], + description: 'Output format.', + default: 'json', + }, + shape: { + type: 'string', + enum: ['auto', 'object', 'array'], + description: 'Output shape for JSON (mirrors input by default).', + default: 'auto', + }, + map: { + type: 'string', + description: + 'Map record fields to canonical keys: field=sourceKey entries separated by commas, ' + + "e.g. 'time=timestamp,volume=amount'.", + }, + output_path: { + type: 'string', + description: + 'Optional output file path (csv, json, jsonl). If omitted, output is returned as text.', + }, + }, + required: ['input_path'], + }, + }, + { + name: 'audit_ohlcv_file', + description: + 'Audit a local OHLCV file and report whether its data is trustworthy, and ' + + 'exactly why. Accepts CSV, JSON, JSONL, NDJSON, or Parquet input. Reports ' + + 'record count, time range, source timeframe, ordering (out-of-order count ' + + 'and max lateness), duplicate timestamps, OHLC integrity violations, bad ' + + 'values (NaN/Infinity/negative prices/volume), and missing bars. Use this ' + + 'before resampling to check the source is sane, and to learn the base ' + + 'timeframe to pass to resample_ohlcv_file.', + inputSchema: { + type: 'object', + properties: { + input_path: { + type: 'string', + description: 'Input file path (csv, json, jsonl, ndjson, parquet).', + }, + map: { + type: 'string', + description: + 'Map record fields to canonical keys: field=sourceKey entries separated by commas, ' + + "e.g. 'time=timestamp,volume=amount'.", + }, + }, + required: ['input_path'], + }, + }, +] as const; + +class RpcError extends Error { + constructor(readonly code: number, message: string) { + super(message); + } +} + +function writeMessage(msg: unknown): void { + process.stdout.write(JSON.stringify(msg) + '\n'); +} + +function toolResult(text: string, isError = false) { + return { content: [{ type: 'text', text }], isError }; +} + +async function callTool(params: any): Promise { + const name = params?.name; + const args = params?.arguments ?? {}; + + if (name !== 'resample_ohlcv_file' && name !== 'audit_ohlcv_file') { + throw new RpcError(-32602, `Unknown tool: ${name}`); + } + const inputPath = args.input_path; + if (typeof inputPath !== 'string' || inputPath.length === 0) { + throw new RpcError(-32602, 'input_path is required and must be a non-empty string'); + } + + // Build the exact argv the `ohlc` CLI would receive, then run it in-process. + const argv = + name === 'audit_ohlcv_file' + ? [ + process.argv[0] ?? 'node', + process.argv[1] ?? 'mcp', + '-i', String(inputPath), + '--audit', + ] + : [ + process.argv[0] ?? 'node', + process.argv[1] ?? 'mcp', + '-i', String(inputPath), + '-b', String(args.base_timeframe ?? 60), + '-n', String(args.new_timeframe ?? 300), + '-f', String(args.format ?? 'json'), + '-s', String(args.shape ?? 'auto'), + ]; + if (args.map !== undefined) argv.push('--map', String(args.map)); + if (args.output_path !== undefined && name !== 'audit_ohlcv_file') { + argv.push('-o', String(args.output_path)); + } + + const stdout = new CaptureStream(); + const stderr = new CaptureStream(); + try { + await runCli(argv, emptyReadable(), stdout, stderr); + } catch (err) { + return toolResult(err instanceof Error ? err.message : String(err), true); + } + + const errText = stderr.text; + if (errText) { + return toolResult(errText, true); + } + const outText = stdout.text; + if (args.output_path !== undefined) { + return toolResult(`Wrote resampled output to ${args.output_path}`); + } + if (!outText) { + return toolResult('No output was produced.', true); + } + return toolResult(outText); +} + +/** Dispatch a JSON-RPC request to its handler and return the result object. */ +export function dispatch(msg: any): Promise { + switch (msg.method) { + case 'initialize': + return Promise.resolve({ + protocolVersion: msg.params?.protocolVersion ?? DEFAULT_PROTOCOL_VERSION, + capabilities: { tools: {} }, + serverInfo: { name: SERVER_NAME, version: VERSION }, + }); + case 'tools/list': + return Promise.resolve({ tools: TOOLS }); + case 'tools/call': + return callTool(msg.params); + case 'ping': + return Promise.resolve(null); + default: + return Promise.reject(new RpcError(-32601, `Method not found: ${msg.method}`)); + } +} + +async function main(): Promise { + const rl = readline.createInterface({ input: process.stdin, crlfDelay: Infinity }); + for await (const line of rl) { + if (!line) continue; + let msg: any; + try { + msg = JSON.parse(line); + } catch { + continue; // not JSON-RPC; ignore + } + // Notifications have no id and get no response. + if (msg.id === undefined || msg.id === null) continue; + try { + const result = await dispatch(msg); + writeMessage({ jsonrpc: '2.0', id: msg.id, result }); + } catch (err) { + const e = err instanceof RpcError ? err : new RpcError(-32603, err instanceof Error ? err.message : String(err)); + writeMessage({ jsonrpc: '2.0', id: msg.id, error: { code: e.code, message: e.message } }); + } + } +} + +// ESM equivalent of `require.main === module`: only run the server when this +// file is executed directly (e.g. `node dist/mcp.js`), never when imported. +const isMain = + process.argv[1] && + path.resolve(fileURLToPath(import.meta.url)) === path.resolve(process.argv[1]); +if (isMain) { + main(); +}