From a45b8da1f72c3071500946672bc15d781b4b2d34 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 18 Aug 2026 14:10:41 +0000 Subject: [PATCH] feature(Files): Download one file per day for the daily parquet layout MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit From 2026-08-01 the API returns one parquet per day rather than one per month, and `FileInfo` gains an optional `day`. The CLI discarded that field, so every daily file in a month resolved to the same output name: 2026-08-01 ─┐ 2026-08-02 ─┼─> outputDir/2026-08.parquet ... ─┘ Up to `--max-concurrent` goroutines then raced to `os.Create` and copy into that one path. A 31-day August left a single file holding one arbitrary day, or an interleave of several, and the CLI still exited 0 reporting "Successfully downloaded 31 files". Daily files are now named `2026-08-01.parquet`; monthly files keep `2026-08.parquet`. Both parts are zero-padded so a listing sorts by date — that is a local naming choice, not the R2 key format, which leaves the month unpadded. `Day` is a `*int` rather than an `int`, so an absent field stays distinct from a literal 0. Reading a stray 0 as "monthly" would put a daily file back on the monthly file's path. Names are resolved and checked for duplicates before anything is written, so a response that somehow collides fails the run instead of silently letting the last writer win. That guard, not the naming, is what makes this class of bug impossible to reintroduce. Also splits one attempt out of `downloadToFile`'s retry loop as `downloadOnce`. The loop body used `defer` for both the response body and the destination file, and a deferred call runs at function return, not at the end of the iteration — so every attempt's handles stayed open for the whole retry sequence and a failed attempt could flush into a path a later attempt had already truncated. Same corrupt-output symptom, same function, so it is fixed here rather than left behind. Tests: 10 new cases in daily_files_test.go, including an end-to-end run against a stub API asserting three days produce three files with three distinct bodies. Reverting the naming fails four of them. The five integration tests that need APERIODIC_API_KEY fail identically on main without it. gofmt also realigned the pre-existing `Exchange` const block in types.go. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01YW171yVjpD7c59PtLT4zKz --- README.md | 12 +- daily_files_test.go | 352 ++++++++++++++++++++++++++++++++++++++++++++ processor.go | 107 +++++++++++--- types.go | 12 +- 4 files changed, 456 insertions(+), 27 deletions(-) create mode 100644 daily_files_test.go diff --git a/README.md b/README.md index 3c88638..a23e7f2 100644 --- a/README.md +++ b/README.md @@ -177,7 +177,17 @@ aperiodic ohlcv --preview \ ## Output -All data commands download **Parquet files** to `--output-dir`. Files are fetched concurrently (tunable via `--max-concurrent`) and named by year and month. +All data commands download **Parquet files** to `--output-dir`, fetched concurrently (tunable via `--max-concurrent`). + +History up to **2026-07-31** is one file per month; from **2026-08-01** onwards it is one file per day. You always ask for a date range and get back every file covering it, so a range spanning the changeover downloads the earlier months as monthly files followed by a daily file per day. + +Filenames follow the granularity, zero-padded so a directory listing sorts chronologically: + +``` +2026-07.parquet # monthly +2026-08-01.parquet # daily +2026-08-02.parquet +``` ## Build from Source diff --git a/daily_files_test.go b/daily_files_test.go new file mode 100644 index 0000000..e8dd8a0 --- /dev/null +++ b/daily_files_test.go @@ -0,0 +1,352 @@ +package aperiodic + +// Tests for the daily-parquet layout. +// +// Data up to 2026-07-31 is served as one parquet per month; from 2026-08-01 the +// API returns one per day, so a single response can hold many files sharing the +// same year and month. These tests pin the property that makes that safe: every +// file in a response lands on its own path on disk. + +import ( + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "slices" + "strings" + "testing" +) + +func dayPtr(d int) *int { return &d } + +func TestParquetFilename(t *testing.T) { + // The month is zero-padded here even though the R2 object key writes it + // unpadded. These are local filenames chosen for sortability, not keys — + // do not "align" them with the producer's key format. + tests := []struct { + name string + file FileInfo + expected string + }{ + {"monthly file omits the day", FileInfo{Year: 2026, Month: 7}, "2026-07.parquet"}, + {"daily file appends a zero-padded day", FileInfo{Year: 2026, Month: 8, Day: dayPtr(1)}, "2026-08-01.parquet"}, + {"two-digit day is unchanged", FileInfo{Year: 2026, Month: 8, Day: dayPtr(31)}, "2026-08-31.parquet"}, + {"single-digit month and day both pad", FileInfo{Year: 2027, Month: 3, Day: dayPtr(9)}, "2027-03-09.parquet"}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := parquetFilename(tt.file); got != tt.expected { + t.Errorf("expected %s, got %s", tt.expected, got) + } + }) + } +} + +func TestParquetFilename_SortsChronologically(t *testing.T) { + // The reason for zero-padding: a directory listing is the user's index of + // what they downloaded, so lexical order has to be date order. + files := []FileInfo{ + {Year: 2026, Month: 8, Day: dayPtr(10)}, + {Year: 2026, Month: 8, Day: dayPtr(2)}, + {Year: 2026, Month: 9, Day: dayPtr(1)}, + {Year: 2026, Month: 8, Day: dayPtr(31)}, + } + + names := make([]string, len(files)) + for i, f := range files { + names[i] = parquetFilename(f) + } + slices.Sort(names) + + expected := []string{ + "2026-08-02.parquet", + "2026-08-10.parquet", + "2026-08-31.parquet", + "2026-09-01.parquet", + } + if !slices.Equal(names, expected) { + t.Errorf("expected %v, got %v", expected, names) + } +} + +func TestResolveFilenames_RejectsCollisions(t *testing.T) { + // Two files in the same month with no day is exactly the shape the CLI + // produced before FileInfo carried one: a month of daily downloads all + // pointed at 2026-08.parquet, raced to truncate it, and still exited 0. + files := []FileInfo{ + {Year: 2026, Month: 8, URL: "first"}, + {Year: 2026, Month: 8, URL: "second"}, + } + + _, err := resolveFilenames(files) + if err == nil { + t.Fatal("expected an error when two files resolve to the same name") + } + if !strings.Contains(err.Error(), "2026-08.parquet") { + t.Errorf("expected the colliding name in the error, got: %v", err) + } +} + +func TestResolveFilenames_DailyFilesDoNotCollide(t *testing.T) { + files := make([]FileInfo, 0, 31) + for d := 1; d <= 31; d++ { + files = append(files, FileInfo{Year: 2026, Month: 8, Day: dayPtr(d)}) + } + + names, err := resolveFilenames(files) + if err != nil { + t.Fatalf("expected 31 distinct names, got error: %v", err) + } + + unique := make(map[string]struct{}, len(names)) + for _, n := range names { + unique[n] = struct{}{} + } + if len(unique) != 31 { + t.Errorf("expected 31 distinct names, got %d", len(unique)) + } +} + +func TestResolveFilenames_MonthlyAndDailyCoexist(t *testing.T) { + // A range spanning the cutover: July monthly, then August daily. The + // monthly name must not collide with any daily name. + files := []FileInfo{ + {Year: 2026, Month: 7}, + {Year: 2026, Month: 8, Day: dayPtr(1)}, + {Year: 2026, Month: 8, Day: dayPtr(2)}, + } + + names, err := resolveFilenames(files) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + expected := []string{"2026-07.parquet", "2026-08-01.parquet", "2026-08-02.parquet"} + if !slices.Equal(names, expected) { + t.Errorf("expected %v, got %v", expected, names) + } +} + +func TestFileInfo_DecodesOptionalDay(t *testing.T) { + const body = `{"files":[ + {"year":2026,"month":7,"url":"monthly"}, + {"year":2026,"month":8,"day":1,"url":"daily"} + ]}` + + var resp AggregateDataResponse + if err := json.Unmarshal([]byte(body), &resp); err != nil { + t.Fatalf("failed to decode: %v", err) + } + + if len(resp.Files) != 2 { + t.Fatalf("expected 2 files, got %d", len(resp.Files)) + } + if resp.Files[0].Day != nil { + t.Errorf("expected the monthly file to have no day, got %d", *resp.Files[0].Day) + } + if resp.Files[1].Day == nil { + t.Fatal("expected the daily file to carry a day") + } + if *resp.Files[1].Day != 1 { + t.Errorf("expected day 1, got %d", *resp.Files[1].Day) + } +} + +// serveFiles stands up a stub API returning `files`, with each URL pointing at +// a blob endpoint on the same server whose body names the file it belongs to. +func serveFiles(t *testing.T, build func(baseURL string) []FileInfo) *httptest.Server { + t.Helper() + + var srv *httptest.Server + srv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if name, isBlob := strings.CutPrefix(r.URL.Path, "/blob/"); isBlob { + _, _ = fmt.Fprintf(w, "contents of %s", name) + return + } + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(AggregateDataResponse{Files: build(srv.URL)}) + })) + t.Cleanup(srv.Close) + + return srv +} + +func downloadedNames(t *testing.T, dir string) []string { + t.Helper() + + entries, err := filepath.Glob(filepath.Join(dir, "*.parquet")) + if err != nil { + t.Fatalf("failed to glob output dir: %v", err) + } + + names := make([]string, len(entries)) + for i, e := range entries { + names[i] = filepath.Base(e) + } + slices.Sort(names) + + return names +} + +func TestCLI_DailyFilesWriteOnePerDay(t *testing.T) { + // The regression this whole change exists for. Pre-fix all three days + // resolved to 2026-08.parquet, so three goroutines truncated and rewrote + // one file and the CLI reported three successes. + srv := serveFiles(t, func(baseURL string) []FileInfo { + files := make([]FileInfo, 0, 3) + for d := 1; d <= 3; d++ { + files = append(files, FileInfo{ + Year: 2026, + Month: 8, + Day: dayPtr(d), + URL: fmt.Sprintf("%s/blob/day-%02d", baseURL, d), + }) + } + return files + }) + + t.Setenv("APERIODIC_API_URL", srv.URL) + t.Setenv("APERIODIC_API_KEY", "test-key") + outputDir := t.TempDir() + + stdout, stderr, code := runCLI( + "ohlcv", + "-exchange", "binance-futures", + "-symbol", "perpetual-BTC-USDT:USDT", + "-interval", "1h", + "-start-date", "2026-08-01", + "-end-date", "2026-08-03", + "-output-dir", outputDir, + ) + if code != 0 { + t.Fatalf("expected exit code 0, got %d; stderr: %s", code, stderr) + } + if !strings.Contains(stdout, "Successfully downloaded 3 files") { + t.Errorf("expected three successes, got: %s", stdout) + } + + names := downloadedNames(t, outputDir) + expected := []string{"2026-08-01.parquet", "2026-08-02.parquet", "2026-08-03.parquet"} + if !slices.Equal(names, expected) { + t.Fatalf("expected %v, got %v", expected, names) + } + + // Distinct names alone would still pass if every file held the same bytes, + // so check each one received its own day's body. + for d, name := range names { + body, err := os.ReadFile(filepath.Join(outputDir, name)) + if err != nil { + t.Fatalf("failed to read %s: %v", name, err) + } + want := fmt.Sprintf("contents of day-%02d", d+1) + if string(body) != want { + t.Errorf("expected %s to hold %q, got %q", name, want, string(body)) + } + } +} + +func TestCLI_RangeAcrossTheCutover(t *testing.T) { + // July as a monthly file, August as daily ones — the shape a real query + // spanning 2026-08-01 returns. + srv := serveFiles(t, func(baseURL string) []FileInfo { + return []FileInfo{ + {Year: 2026, Month: 7, URL: baseURL + "/blob/july"}, + {Year: 2026, Month: 8, Day: dayPtr(1), URL: baseURL + "/blob/aug-01"}, + {Year: 2026, Month: 8, Day: dayPtr(2), URL: baseURL + "/blob/aug-02"}, + } + }) + + t.Setenv("APERIODIC_API_URL", srv.URL) + t.Setenv("APERIODIC_API_KEY", "test-key") + outputDir := t.TempDir() + + _, stderr, code := runCLI( + "ohlcv", + "-exchange", "binance-futures", + "-symbol", "perpetual-BTC-USDT:USDT", + "-interval", "1h", + "-start-date", "2026-07-30", + "-end-date", "2026-08-02", + "-output-dir", outputDir, + ) + if code != 0 { + t.Fatalf("expected exit code 0, got %d; stderr: %s", code, stderr) + } + + names := downloadedNames(t, outputDir) + expected := []string{"2026-07.parquet", "2026-08-01.parquet", "2026-08-02.parquet"} + if !slices.Equal(names, expected) { + t.Errorf("expected %v, got %v", expected, names) + } +} + +func TestCLI_MonthlyOnlyRangeIsUnchanged(t *testing.T) { + srv := serveFiles(t, func(baseURL string) []FileInfo { + return []FileInfo{ + {Year: 2025, Month: 1, URL: baseURL + "/blob/jan"}, + {Year: 2025, Month: 2, URL: baseURL + "/blob/feb"}, + } + }) + + t.Setenv("APERIODIC_API_URL", srv.URL) + t.Setenv("APERIODIC_API_KEY", "test-key") + outputDir := t.TempDir() + + _, stderr, code := runCLI( + "ohlcv", + "-exchange", "binance-futures", + "-symbol", "perpetual-BTC-USDT:USDT", + "-interval", "1h", + "-start-date", "2025-01-01", + "-end-date", "2025-02-28", + "-output-dir", outputDir, + ) + if code != 0 { + t.Fatalf("expected exit code 0, got %d; stderr: %s", code, stderr) + } + + names := downloadedNames(t, outputDir) + expected := []string{"2025-01.parquet", "2025-02.parquet"} + if !slices.Equal(names, expected) { + t.Errorf("expected %v, got %v", expected, names) + } +} + +func TestCLI_CollidingFilesFailTheRun(t *testing.T) { + // An API that omitted `day` on daily files would put the CLI back in the + // silent-overwrite state. It has to fail instead of writing whichever + // download finishes last. + srv := serveFiles(t, func(baseURL string) []FileInfo { + return []FileInfo{ + {Year: 2026, Month: 8, URL: baseURL + "/blob/first"}, + {Year: 2026, Month: 8, URL: baseURL + "/blob/second"}, + } + }) + + t.Setenv("APERIODIC_API_URL", srv.URL) + t.Setenv("APERIODIC_API_KEY", "test-key") + outputDir := t.TempDir() + + _, stderr, code := runCLI( + "ohlcv", + "-exchange", "binance-futures", + "-symbol", "perpetual-BTC-USDT:USDT", + "-interval", "1h", + "-start-date", "2026-08-01", + "-end-date", "2026-08-02", + "-output-dir", outputDir, + ) + if code != 1 { + t.Fatalf("expected exit code 1, got %d; stderr: %s", code, stderr) + } + if !strings.Contains(stderr, "2026-08.parquet") { + t.Errorf("expected the colliding name in stderr, got: %s", stderr) + } + + if names := downloadedNames(t, outputDir); len(names) != 0 { + t.Errorf("expected nothing written when names collide, got %v", names) + } +} diff --git a/processor.go b/processor.go index d32484e..773c4fa 100644 --- a/processor.go +++ b/processor.go @@ -13,14 +13,58 @@ import ( type DownloadedFile struct { Year int Month int + Day *int Filename string } +// parquetFilename is the on-disk name for one downloaded object. +// +// Both parts are zero-padded so a directory listing sorts chronologically. That +// is a choice about local filenames, not a mirror of the R2 key, which writes +// the month unpadded — do not "align" the two. +func parquetFilename(f FileInfo) string { + if f.Day == nil { + return fmt.Sprintf("%d-%02d.parquet", f.Year, f.Month) + } + return fmt.Sprintf("%d-%02d-%02d.parquet", f.Year, f.Month, *f.Day) +} + +// resolveFilenames names every file and rejects the set if two share a name. +// +// Load-bearing, not defensive tidiness: before FileInfo carried a day, every +// daily file in a month resolved to the same name, so a month of downloads +// raced to truncate and rewrite a single path — and the CLI still exited 0 +// reporting one success per file. Colliding names must fail the run rather than +// let whichever goroutine finishes last decide the contents. +func resolveFilenames(files []FileInfo) ([]string, error) { + filenames := make([]string, len(files)) + seen := make(map[string]int, len(files)) + + for i, f := range files { + name := parquetFilename(f) + if first, duplicate := seen[name]; duplicate { + return nil, fmt.Errorf( + "refusing to download: files %d and %d both resolve to %s", + first, i, name, + ) + } + seen[name] = i + filenames[i] = name + } + + return filenames, nil +} + func (c *AperiodicClient) DownloadFilesConcurrently(files []FileInfo, maxConcurrent int, outputDir string) ([]DownloadedFile, error) { if len(files) == 0 { return nil, nil } + filenames, err := resolveFilenames(files) + if err != nil { + return nil, err + } + if err := os.MkdirAll(outputDir, 0755); err != nil { return nil, fmt.Errorf("failed to create output directory: %w", err) } @@ -37,16 +81,17 @@ func (c *AperiodicClient) DownloadFilesConcurrently(files []FileInfo, maxConcurr semaphore <- struct{}{} defer func() { <-semaphore }() - filename := fmt.Sprintf("%d-%02d.parquet", f.Year, f.Month) + filename := filenames[i] destPath := filepath.Join(outputDir, filename) if err := c.downloadToFile(f.URL, destPath, 3); err != nil { - errs[i] = fmt.Errorf("failed to download %d-%02d: %w", f.Year, f.Month, err) + errs[i] = fmt.Errorf("failed to download %s: %w", filename, err) return } results[i] = DownloadedFile{ Year: f.Year, Month: f.Month, + Day: f.Day, Filename: filename, } }(i, file) @@ -70,31 +115,47 @@ func (c *AperiodicClient) downloadToFile(url, destPath string, maxRetries int) e time.Sleep(time.Duration(i) * time.Second) // Simple backoff } - resp, err := c.HTTPClient.Get(url) - if err != nil { - lastErr = err - continue + retryable, err := c.downloadOnce(url, destPath) + if err == nil { + return nil } - defer resp.Body.Close() - - if resp.StatusCode != http.StatusOK { - lastErr = fmt.Errorf("bad status: %s", resp.Status) - continue + if !retryable { + return err } + lastErr = err + } + return lastErr +} - out, err := os.Create(destPath) - if err != nil { - return fmt.Errorf("failed to create file: %w", err) - } - defer out.Close() +// downloadOnce makes a single attempt, closing the response body and the +// destination file before it returns. It reports whether the failure is worth +// another attempt: transport and HTTP errors are, a local filesystem error is +// not. +// +// Living outside downloadToFile's loop is the point. A `defer` in a loop body +// runs only when the whole function returns, so every attempt's handles stayed +// open for the length of the retry sequence — leaving a failed attempt free to +// flush into a path a later attempt had already truncated. +func (c *AperiodicClient) downloadOnce(url, destPath string) (retryable bool, err error) { + resp, err := c.HTTPClient.Get(url) + if err != nil { + return true, err + } + defer resp.Body.Close() - _, err = io.Copy(out, resp.Body) - if err != nil { - lastErr = err - continue - } + if resp.StatusCode != http.StatusOK { + return true, fmt.Errorf("bad status: %s", resp.Status) + } - return nil + out, err := os.Create(destPath) + if err != nil { + return false, fmt.Errorf("failed to create file: %w", err) } - return lastErr + + _, copyErr := io.Copy(out, resp.Body) + closeErr := out.Close() + if copyErr != nil { + return true, copyErr + } + return true, closeErr } diff --git a/types.go b/types.go index 4ef7a38..856836f 100644 --- a/types.go +++ b/types.go @@ -23,9 +23,9 @@ const ( type Exchange string const ( - ExchangeBinanceFutures Exchange = "binance-futures" - ExchangeOkxPerps Exchange = "okx-perps" - ExchangeHyperliquidPerps Exchange = "hyperliquid-perps" + ExchangeBinanceFutures Exchange = "binance-futures" + ExchangeOkxPerps Exchange = "okx-perps" + ExchangeHyperliquidPerps Exchange = "hyperliquid-perps" ) type TradeMetric string @@ -70,6 +70,12 @@ type FileInfo struct { Year int `json:"year"` Month int `json:"month"` URL string `json:"url"` + + // Day is set only on daily files, which cover 2026-08-01 onwards; monthly + // files omit the field. A pointer rather than a plain int so an absent day + // stays distinguishable from a literal 0 — reading a stray 0 as "monthly" + // would collide with the month's real monthly file. + Day *int `json:"day,omitempty"` } type AggregateDataResponse struct {