From 5385c51638770023a1ed678664be898ab0e95201 Mon Sep 17 00:00:00 2001
From: Noobb-byte <316570727+Noobb-byte@users.noreply.github.com>
Date: Fri, 16 Oct 2026 22:04:59 +0100
Subject: [PATCH] feat(ingest): add historical backfill CLI and persistent gap
 detection (#241)

- Add `IngestionGap` model to track gaps per network, venue/source, and pair
  with statuses 'open', 'filled', and 'permanent'.
- Implement idempotent gap detection with cadence-based hole detection and
  automatic merging of overlapping/adjacent intervals.
- Add backfill service and CLI script (`npm run backfill`) to refetch trade
  history via Horizon for SDEX and AMM, writing through the idempotent
  `upsertPricePoints` ingestion path.
- Mark gaps for non-history snapshot sources (Soroswap, Aquarius) as 'permanent'
  without failing noisily.
- Declare `incomplete: true` and overlapping `gaps` in time-window API responses
  (`/prices/history`, `/candles/*`, `/price/*/*history`, `/price/twap/*`,
  `/price/vwap/*`) when windows overlap open or permanent gaps.
- Update OpenAPI specifications (`openapi.yaml`, `openapi.json`), route schemas,
  README documentation, and add test suite covering detection, backfill, and API responses.
---
 README.md                       |  46 +++
 openapi.json                    |  75 ++++
 openapi.yaml                    |  47 +++
 package.json                    |   3 +-
 prisma/schema.prisma            |  17 +
 scripts/backfill.ts             | 136 +++++++
 src/__tests__/backfill.test.ts  | 608 ++++++++++++++++++++++++++++++++
 src/api/history.ts              |  10 +
 src/api/rest.ts                 |  22 ++
 src/api/schemas.ts              |  15 +
 src/backfill/backfillService.ts | 302 ++++++++++++++++
 src/backfill/gaps.ts            | 443 +++++++++++++++++++++++
 src/backfill/types.ts           |  40 +++
 src/db.ts                       |   4 +
 src/routes/candles.ts           |  13 +
 src/routes/price.ts             |  24 ++
 16 files changed, 1804 insertions(+), 1 deletion(-)
 create mode 100644 scripts/backfill.ts
 create mode 100644 src/__tests__/backfill.test.ts
 create mode 100644 src/backfill/backfillService.ts
 create mode 100644 src/backfill/gaps.ts
 create mode 100644 src/backfill/types.ts

diff --git a/README.md b/README.md
index 2201d343..73df3e3c 100644
--- a/README.md
+++ b/README.md
@@ -483,6 +483,52 @@ curl -s http://localhost:3002/price/XLM/USDC | jq '{price, ammPrice, lastUpdated
 
 *(Note: timestamp values above reflect the dynamic hourly anchor at seed execution time and request time).*
 
+## Backfill & Ingestion Gap Detection
+
+Lens provides persistent gap detection and an idempotent historical backfill CLI to recover missing trade history and declare incomplete time windows in API responses.
+
+### 1. Gap Detection & IngestionGap Model
+Gaps are tracked per `(network, source, pair)` in the `IngestionGap` table:
+- `status`: `'open'` (detected hole in history), `'filled'` (successfully backfilled), or `'permanent'` (data cannot be backfilled).
+- Gaps are detected when the elapsed time between consecutive price points exceeds the expected ingestion cadence (default 5 minutes).
+- **Idempotency & Merging:** Re-running gap detection never duplicates records; overlapping or adjacent intervals are merged automatically into single continuous ranges.
+
+### 2. Backfill CLI
+Run the backfill script for historical sources:
+
+```bash
+# Backfill SDEX trades for XLM/USDC
+npm run backfill -- --source SDEX --pair XLM/USDC --start 2026-10-15T00:00:00Z --end 2026-10-15T06:00:00Z --network testnet
+
+# Detect gaps and backfill them automatically
+npm run backfill -- --source SDEX --pair XLM/USDC --start 2026-10-15T00:00:00Z --end 2026-10-15T06:00:00Z --detect
+```
+
+- **History Support:**
+  - `SDEX` and `AMM`: Supported via Horizon historical trade paging. Refetched points are written through the same idempotent `upsertPricePoints` path as live ingestion. Gaps are marked `filled`.
+  - `soroswap_amm` and `aquarius_amm`: State-only venues that do not retain historical trade logs. Attempting to backfill them cleanly marks the gap as `permanent` and exits without error.
+- **Idempotent Re-runs:** Running the same backfill range a second time produces 0 duplicate records and leaves the gap marked `filled`.
+
+### 3. Incomplete Window Declarations in API Responses
+When querying time-window endpoints (such as `/prices/history`, `/candles/:assetA/:assetB`, `/price/:assetA/:assetB/history`, `/price/twap/*`, and `/price/vwap/*`):
+- If the requested window overlaps any `open` or `permanent` gaps for that pair and network, the response includes `incomplete: true` along with the list of overlapping `gaps`:
+  ```json
+  {
+    "pair": "XLM/USDC",
+    "interval": "1m",
+    "incomplete": true,
+    "gaps": [
+      {
+        "source": "SDEX",
+        "start": "2026-10-15T01:00:00.000Z",
+        "end": "2026-10-15T03:00:00.000Z",
+        "status": "open"
+      }
+    ],
+    "points": [...]
+  }
+  ```
+- Windows that are clean or cover only `filled` gaps omit the `incomplete` and `gaps` fields.
 
 ## Environment Variables
 
diff --git a/openapi.json b/openapi.json
index b0576e01..ae3793e7 100644
--- a/openapi.json
+++ b/openapi.json
@@ -250,6 +250,15 @@
                     "window": {
                       "type": "string"
                     },
+                    "incomplete": {
+                      "type": "boolean"
+                    },
+                    "gaps": {
+                      "type": "array",
+                      "items": {
+                        "$ref": "#/components/schemas/DataGap"
+                      }
+                    },
                     "buckets": {
                       "type": "array",
                       "items": {
@@ -421,6 +430,15 @@
                     "to": {
                       "type": "string"
                     },
+                    "incomplete": {
+                      "type": "boolean"
+                    },
+                    "gaps": {
+                      "type": "array",
+                      "items": {
+                        "$ref": "#/components/schemas/DataGap"
+                      }
+                    },
                     "candles": {
                       "type": "array",
                       "items": {
@@ -550,6 +568,15 @@
                       "type": "string",
                       "format": "date-time"
                     },
+                    "incomplete": {
+                      "type": "boolean"
+                    },
+                    "gaps": {
+                      "type": "array",
+                      "items": {
+                        "$ref": "#/components/schemas/DataGap"
+                      }
+                    },
                     "count": {
                       "type": "integer"
                     },
@@ -2159,6 +2186,15 @@
               "iqr",
               "modified_zscore"
             ]
+          },
+          "incomplete": {
+            "type": "boolean"
+          },
+          "gaps": {
+            "type": "array",
+            "items": {
+              "$ref": "#/components/schemas/DataGap"
+            }
           }
         }
       },
@@ -2221,6 +2257,45 @@
               "iqr",
               "modified_zscore"
             ]
+          },
+          "incomplete": {
+            "type": "boolean"
+          },
+          "gaps": {
+            "type": "array",
+            "items": {
+              "$ref": "#/components/schemas/DataGap"
+            }
+          }
+        }
+      },
+      "DataGap": {
+        "type": "object",
+        "required": [
+          "source",
+          "start",
+          "end",
+          "status"
+        ],
+        "additionalProperties": false,
+        "properties": {
+          "source": {
+            "type": "string"
+          },
+          "start": {
+            "type": "string",
+            "format": "date-time"
+          },
+          "end": {
+            "type": "string",
+            "format": "date-time"
+          },
+          "status": {
+            "type": "string",
+            "enum": [
+              "open",
+              "permanent"
+            ]
           }
         }
       },
diff --git a/openapi.yaml b/openapi.yaml
index 182db46d..5cd723c6 100644
--- a/openapi.yaml
+++ b/openapi.yaml
@@ -161,6 +161,12 @@ paths:
                     type: string
                   window:
                     type: string
+                  incomplete:
+                    type: boolean
+                  gaps:
+                    type: array
+                    items:
+                      $ref: '#/components/schemas/DataGap'
                   buckets:
                     type: array
                     items:
@@ -265,6 +271,12 @@ paths:
                     type: string
                   to:
                     type: string
+                  incomplete:
+                    type: boolean
+                  gaps:
+                    type: array
+                    items:
+                      $ref: '#/components/schemas/DataGap'
                   candles:
                     type: array
                     items:
@@ -348,6 +360,12 @@ paths:
                   to:
                     type: string
                     format: date-time
+                  incomplete:
+                    type: boolean
+                  gaps:
+                    type: array
+                    items:
+                      $ref: '#/components/schemas/DataGap'
                   count:
                     type: integer
                   points:
@@ -1378,6 +1396,12 @@ components:
         filterMethod:
           type: string
           enum: [iqr, modified_zscore]
+        incomplete:
+          type: boolean
+        gaps:
+          type: array
+          items:
+            $ref: '#/components/schemas/DataGap'
 
     VwapResponse:
       type: object
@@ -1411,6 +1435,29 @@ components:
         filterMethod:
           type: string
           enum: [iqr, modified_zscore]
+        incomplete:
+          type: boolean
+        gaps:
+          type: array
+          items:
+            $ref: '#/components/schemas/DataGap'
+
+    DataGap:
+      type: object
+      required: [source, start, end, status]
+      additionalProperties: false
+      properties:
+        source:
+          type: string
+        start:
+          type: string
+          format: date-time
+        end:
+          type: string
+          format: date-time
+        status:
+          type: string
+          enum: [open, permanent]
 
     VolumeResponse:
       type: object
diff --git a/package.json b/package.json
index c25cfd9e..9e1ac839 100644
--- a/package.json
+++ b/package.json
@@ -19,7 +19,8 @@
     "key:issue": "tsx scripts/issue-api-key.ts",
     "db:push": "prisma db push",
     "db:generate": "prisma generate",
-    "seed": "tsx scripts/seed.ts"
+    "seed": "tsx scripts/seed.ts",
+    "backfill": "tsx scripts/backfill.ts"
   },
   "dependencies": {
     "@fastify/compress": "^8.3.1",
diff --git a/prisma/schema.prisma b/prisma/schema.prisma
index 29189728..4c1de015 100644
--- a/prisma/schema.prisma
+++ b/prisma/schema.prisma
@@ -270,3 +270,20 @@ model ApiKey {
 
   @@map("api_keys")
 }
+
+model IngestionGap {
+  id         String   @id @default(uuid())
+  network    String   @default("testnet")
+  source     String
+  pair       String   @map("pair_key")
+  start      DateTime @map("start_time")
+  end        DateTime @map("end_time")
+  status     String   @default("open")
+  detectedAt DateTime @default(now()) @map("detected_at")
+  updatedAt  DateTime @default(now()) @updatedAt @map("updated_at")
+
+  @@index([network, pair, source, status])
+  @@index([network, pair, start, end])
+  @@map("ingestion_gaps")
+}
+
diff --git a/scripts/backfill.ts b/scripts/backfill.ts
new file mode 100644
index 00000000..4dd0856e
--- /dev/null
+++ b/scripts/backfill.ts
@@ -0,0 +1,136 @@
+/**
+ * CLI command to backfill data for a given source, pair, and time range.
+ *
+ * Usage:
+ *   npm run backfill -- --source SDEX --pair XLM/USDC --start 2026-10-15T00:00:00Z --end 2026-10-15T06:00:00Z
+ *   npm run backfill -- --detect --source SDEX --pair XLM/USDC
+ *
+ * Idempotent: re-running against the same range refetches and writes through
+ * the same idempotent live ingestion path (upsertPricePoints), producing zero duplicates.
+ */
+import 'dotenv/config'
+import { activeNetwork, type NetworkName } from '../src/config'
+import { backfill } from '../src/backfill/backfillService'
+import { detectGaps } from '../src/backfill/gaps'
+import { prisma, pgPool } from '../src/db'
+
+function parseArgs(argv: string[]): Record<string, string> {
+  const args: Record<string, string> = {}
+  for (let i = 0; i < argv.length; i++) {
+    const arg = argv[i]
+    if (arg.startsWith('--')) {
+      const key = arg.slice(2)
+      const next = argv[i + 1]
+      if (next && !next.startsWith('--')) {
+        args[key] = next
+        i++
+      } else {
+        args[key] = 'true'
+      }
+    }
+  }
+  return args
+}
+
+function printUsage() {
+  console.log(`
+Usage:
+  npm run backfill -- --source <SOURCE> --pair <PAIR> --start <ISO_DATE> --end <ISO_DATE> [--network <NETWORK>]
+
+Options:
+  --source   Source venue (e.g. SDEX, AMM, soroswap_amm, aquarius_amm)
+  --pair     Watched pair (e.g. XLM/USDC or USDC:GBBD.../XLM)
+  --start    Start timestamp (ISO 8601 string)
+  --end      End timestamp (ISO 8601 string)
+  --network  Stellar network (testnet | mainnet, default: activeNetwork)
+  --detect   Run gap detection for the pair/source first
+  --cadence  Expected cadence threshold in milliseconds (optional)
+`)
+}
+
+async function main() {
+  const args = parseArgs(process.argv.slice(2))
+
+  if (args.help || (!args.source && !args.pair && !args.start)) {
+    printUsage()
+    process.exit(args.help ? 0 : 1)
+  }
+
+  const network: NetworkName = (args.network as NetworkName) ?? activeNetwork
+  const source = args.source
+  const pair = args.pair
+
+  if (!source) {
+    console.error('Error: --source is required (e.g. --source SDEX)')
+    process.exit(1)
+  }
+  if (!pair) {
+    console.error('Error: --pair is required (e.g. --pair XLM/USDC)')
+    process.exit(1)
+  }
+
+  try {
+    // If --detect flag is set, run detection
+    if (args.detect === 'true') {
+      console.log(`[gaps:detect] Detecting gaps for source=${source}, pair=${pair}, network=${network}…`)
+      const cadenceMs = args.cadence ? parseInt(args.cadence, 10) : undefined
+      const start = args.start ? new Date(args.start) : undefined
+      const end = args.end ? new Date(args.end) : undefined
+
+      const detected = await detectGaps({
+        network,
+        pair,
+        source,
+        start,
+        end,
+        cadenceMs,
+      })
+
+      console.log(`[gaps:detect] Found ${detected.length} gap record(s):`)
+      for (const g of detected) {
+        console.log(`   - [${g.start.toISOString()} → ${g.end.toISOString()}] status: ${g.status}`)
+      }
+
+      if (!args.start && !args.end) {
+        // Detection only
+        return
+      }
+    }
+
+    if (!args.start || !args.end) {
+      console.error('Error: --start and --end are required for backfill (e.g. --start 2026-10-15T00:00:00Z --end 2026-10-15T06:00:00Z)')
+      process.exit(1)
+    }
+
+    const start = new Date(args.start)
+    const end = new Date(args.end)
+
+    console.log(`[backfill] Running backfill for source=${source}, pair=${pair}, network=${network}`)
+    console.log(`   Window: ${start.toISOString()} → ${end.toISOString()}`)
+
+    const result = await backfill({
+      source,
+      pair,
+      start,
+      end,
+      network,
+    })
+
+    if (result.unfillable) {
+      console.warn(`⚠️  ${result.message}`)
+      console.log('   Result: Unfillable gap recorded with status=permanent.')
+    } else {
+      console.log('✅ Backfill complete.')
+      console.log(`   Points ingested: ${result.pointsIngested}`)
+      console.log(`   Gaps updated:    ${result.gapsFilled}`)
+    }
+  } finally {
+    await prisma.$disconnect()
+    await pgPool.end()
+  }
+}
+
+main().catch((err) => {
+  console.error('[backfill] Fatal error:', (err as Error).message)
+  process.exit(1)
+})
diff --git a/src/__tests__/backfill.test.ts b/src/__tests__/backfill.test.ts
new file mode 100644
index 00000000..109239c5
--- /dev/null
+++ b/src/__tests__/backfill.test.ts
@@ -0,0 +1,608 @@
+import { describe, it, expect, vi, beforeEach } from 'vitest'
+import Fastify from 'fastify'
+
+const { mockPrisma, mockPgPool, mockUpsertPricePoints, storedGaps, storedPoints } = vi.hoisted(() => {
+  interface StoredGap {
+    id: string
+    network: string
+    source: string
+    pair: string
+    start: Date
+    end: Date
+    status: string
+    detectedAt: Date
+    updatedAt: Date
+  }
+
+  const storedGaps: StoredGap[] = []
+  const storedPoints: any[] = []
+
+  const mockPrisma = {
+    ingestionGap: {
+      findMany: vi.fn(async (args?: any) => {
+        let filtered = [...storedGaps]
+        if (args?.where) {
+          const { network, pair, source, status, start, end, AND } = args.where
+          if (network) filtered = filtered.filter(g => g.network === network)
+          if (pair) {
+            if (typeof pair === 'string') filtered = filtered.filter(g => g.pair === pair)
+            else if (pair.in) filtered = filtered.filter(g => pair.in.includes(g.pair))
+          }
+          if (source) filtered = filtered.filter(g => g.source === source)
+          if (status) {
+            if (typeof status === 'string') filtered = filtered.filter(g => g.status === status)
+            else if (status.in) filtered = filtered.filter(g => status.in.includes(g.status))
+          }
+          if (start?.lt) filtered = filtered.filter(g => g.start < start.lt)
+          if (end?.gt) filtered = filtered.filter(g => g.end > end.gt)
+          if (AND) {
+            for (const clause of AND) {
+              if (clause.start?.lte) filtered = filtered.filter(g => g.start <= clause.start.lte)
+              if (clause.end?.gte) filtered = filtered.filter(g => g.end >= clause.end.gte)
+            }
+          }
+        }
+        return filtered
+      }),
+      create: vi.fn(async ({ data }: { data: any }) => {
+        const row: StoredGap = {
+          id: `gap-${Date.now()}-${Math.random().toString(36).slice(2, 7)}`,
+          network: data.network,
+          source: data.source,
+          pair: data.pair,
+          start: data.start,
+          end: data.end,
+          status: data.status,
+          detectedAt: new Date(),
+          updatedAt: new Date(),
+        }
+        storedGaps.push(row)
+        return row
+      }),
+      update: vi.fn(async ({ where, data }: { where: { id: string }; data: any }) => {
+        const idx = storedGaps.findIndex(g => g.id === where.id)
+        if (idx === -1) throw new Error('Not found')
+        storedGaps[idx] = {
+          ...storedGaps[idx],
+          ...data,
+          updatedAt: data.updatedAt ?? new Date(),
+        }
+        return storedGaps[idx]
+      }),
+      deleteMany: vi.fn(async ({ where }: { where: { id: { in: string[] } } }) => {
+        let count = 0
+        for (let i = storedGaps.length - 1; i >= 0; i--) {
+          if (where.id.in.includes(storedGaps[i].id)) {
+            storedGaps.splice(i, 1)
+            count++
+          }
+        }
+        return { count }
+      }),
+    },
+    pricePoint: {
+      createMany: vi.fn(async ({ data, skipDuplicates }: { data: any[]; skipDuplicates?: boolean }) => {
+        let inserted = 0
+        for (const item of data) {
+          const id = item.id ?? item.eventId
+          const exists = storedPoints.some(p => p.id === id && p.timestamp.getTime() === item.timestamp.getTime())
+          if (skipDuplicates && exists) {
+            continue
+          }
+          storedPoints.push({ ...item, id })
+          inserted++
+        }
+        return { count: inserted }
+      }),
+    },
+  }
+
+  const mockPgPool = {
+    query: vi.fn(async (sql: string, params: any[]) => {
+      if (sql.includes('MIN(timestamp)')) {
+        const earliest = storedPoints.length > 0 ? storedPoints[0].timestamp : null
+        return { rows: [{ earliest }] }
+      }
+      if (sql.includes('FROM price_snapshots')) {
+        return { rows: [] }
+      }
+      if (sql.includes('FROM price_points')) {
+        if (sql.includes('AS open') || sql.includes('AS time') || sql.includes('COUNT(')) {
+          return { rows: [] }
+        }
+        const filtered = storedPoints
+          .filter(p => p.timestamp >= params[3] && p.timestamp <= params[4])
+          .sort((a, b) => a.timestamp.getTime() - b.timestamp.getTime())
+        return { rows: filtered }
+      }
+      return { rows: [] }
+    }),
+  }
+
+  const mockUpsertPricePoints = vi.fn(async (points: any[], network: any) => {
+    if (points.length === 0) return 0
+    const res = await mockPrisma.pricePoint.createMany({
+      data: points.map(p => ({
+        id: p.id ?? (p.eventId ? `${network}:${p.source}:${p.eventId}` : undefined),
+        network,
+        assetA: p.assetA,
+        assetB: p.assetB,
+        pairKey: p.pairKey,
+        source: p.source,
+        poolId: p.poolId ?? null,
+        price: p.price,
+        baseVolume: p.baseVolume,
+        counterVolume: p.counterVolume,
+        ledger: p.ledger,
+        timestamp: p.timestamp,
+        eventId: p.eventId ?? null,
+      })),
+      skipDuplicates: true,
+    })
+    return res.count
+  })
+
+  return { mockPrisma, mockPgPool, mockUpsertPricePoints, storedGaps, storedPoints }
+})
+
+vi.mock('../db', () => ({
+  prisma: mockPrisma,
+  pgPool: mockPgPool,
+  upsertPricePoints: mockUpsertPricePoints,
+}))
+
+// Attach mocks to globalThis so gaps.ts and db.ts share them cleanly
+;(globalThis as any).__lens_prisma = mockPrisma
+;(globalThis as any).__lens_pgPool = mockPgPool
+
+const mockHorizonTrades = vi.fn()
+vi.mock('../network/clients', () => ({
+  getHorizonServer: () => ({
+    trades: () => ({
+      forAssetPair: () => ({
+        cursor: () => ({
+          limit: () => ({
+            order: () => ({
+              call: mockHorizonTrades,
+            }),
+          }),
+        }),
+      }),
+    }),
+  }),
+  getRpcServer: () => ({}),
+}))
+
+import { detectGaps, recordGaps, findOverlappingGaps, supportsHistory } from '../backfill/gaps'
+import { backfill } from '../backfill/backfillService'
+import { registerPriceRoutes } from '../routes/price'
+import { registerHistoryRoutes } from '../api/history'
+import { registerCandleRoutes } from '../routes/candles'
+import { registerRESTRoutes } from '../api/rest'
+import { registerNetworkSelector } from '../middleware/network'
+
+describe('Gap Detection & Idempotent Recording', () => {
+  beforeEach(() => {
+    storedGaps.length = 0
+    storedPoints.length = 0
+    vi.clearAllMocks()
+    ;(globalThis as any).__lens_prisma = mockPrisma
+    ;(globalThis as any).__lens_pgPool = mockPgPool
+  })
+
+  it('determines source history support correctly', () => {
+    expect(supportsHistory('SDEX')).toBe(true)
+    expect(supportsHistory('sdex')).toBe(true)
+    expect(supportsHistory('AMM')).toBe(true)
+    expect(supportsHistory('soroswap_amm')).toBe(false)
+    expect(supportsHistory('aquarius_amm')).toBe(false)
+    expect(supportsHistory('unknown')).toBe(false)
+  })
+
+  it('detects a missing window when no points exist and records gap with required fields', async () => {
+    const start = new Date('2026-10-15T00:00:00Z')
+    const end = new Date('2026-10-15T06:00:00Z')
+
+    const detected = await detectGaps({
+      network: 'testnet',
+      pair: 'XLM/USDC',
+      source: 'SDEX',
+      start,
+      end,
+      cadenceMs: 300_000,
+    })
+
+    expect(detected).toHaveLength(1)
+    const gap = detected[0]
+    expect(gap.source).toBe('SDEX')
+    expect(gap.pair).toBe('XLM/USDC')
+    expect(gap.start.toISOString()).toBe(start.toISOString())
+    expect(gap.end.toISOString()).toBe(end.toISOString())
+    expect(gap.status).toBe('open')
+    expect(gap.detectedAt).toBeInstanceOf(Date)
+    expect(gap.updatedAt).toBeInstanceOf(Date)
+
+    // Verify stored in persistent storage
+    expect(storedGaps).toHaveLength(1)
+  })
+
+  it('detects internal holes between sparse price points', async () => {
+    const t0 = new Date('2026-10-15T00:00:00Z')
+    const t1 = new Date('2026-10-15T00:02:00Z')
+    // Hole between t1 and t2: 4 hours
+    const t2 = new Date('2026-10-15T04:02:00Z')
+    const t3 = new Date('2026-10-15T04:04:00Z')
+
+    storedPoints.push(
+      { timestamp: t0 },
+      { timestamp: t1 },
+      { timestamp: t2 },
+      { timestamp: t3 },
+    )
+
+    const detected = await detectGaps({
+      network: 'testnet',
+      pair: 'XLM/USDC',
+      source: 'SDEX',
+      start: t0,
+      end: t3,
+      cadenceMs: 300_000, // 5 min
+    })
+
+    expect(detected).toHaveLength(1)
+    expect(detected[0].start.toISOString()).toBe(t1.toISOString())
+    expect(detected[0].end.toISOString()).toBe(t2.toISOString())
+    expect(detected[0].status).toBe('open')
+  })
+
+  it('is idempotent: re-running gap detection produces no duplicates', async () => {
+    const start = new Date('2026-10-15T00:00:00Z')
+    const end = new Date('2026-10-15T06:00:00Z')
+
+    // First run
+    await detectGaps({
+      network: 'testnet',
+      pair: 'XLM/USDC',
+      source: 'SDEX',
+      start,
+      end,
+      cadenceMs: 300_000,
+    })
+    expect(storedGaps).toHaveLength(1)
+    const initialId = storedGaps[0].id
+
+    // Second run with same parameters
+    const secondRun = await detectGaps({
+      network: 'testnet',
+      pair: 'XLM/USDC',
+      source: 'SDEX',
+      start,
+      end,
+      cadenceMs: 300_000,
+    })
+
+    expect(storedGaps).toHaveLength(1)
+    expect(secondRun).toHaveLength(1)
+    expect(secondRun[0].id).toBe(initialId)
+    expect(secondRun[0].start.toISOString()).toBe(start.toISOString())
+    expect(secondRun[0].end.toISOString()).toBe(end.toISOString())
+  })
+
+  it('merges overlapping and adjacent gaps into a single continuous row', async () => {
+    // Gap 1: 00:00 → 02:00
+    await recordGaps({
+      network: 'testnet',
+      pair: 'XLM/USDC',
+      source: 'SDEX',
+      gaps: [{
+        start: new Date('2026-10-15T00:00:00Z'),
+        end: new Date('2026-10-15T02:00:00Z'),
+      }],
+    })
+    expect(storedGaps).toHaveLength(1)
+
+    // Gap 2 overlapping & extending: 01:30 → 04:00
+    await recordGaps({
+      network: 'testnet',
+      pair: 'XLM/USDC',
+      source: 'SDEX',
+      gaps: [{
+        start: new Date('2026-10-15T01:30:00Z'),
+        end: new Date('2026-10-15T04:00:00Z'),
+      }],
+    })
+
+    // Should be merged into one gap: 00:00 → 04:00
+    expect(storedGaps).toHaveLength(1)
+    expect(storedGaps[0].start.toISOString()).toBe('2026-10-15T00:00:00.000Z')
+    expect(storedGaps[0].end.toISOString()).toBe('2026-10-15T04:00:00.000Z')
+  })
+})
+
+describe('Backfill Command & Idempotent Refetch', () => {
+  beforeEach(() => {
+    storedGaps.length = 0
+    storedPoints.length = 0
+    vi.clearAllMocks()
+    ;(globalThis as any).__lens_prisma = mockPrisma
+    ;(globalThis as any).__lens_pgPool = mockPgPool
+  })
+
+  it('validates input ranges (start < end, not in future)', async () => {
+    await expect(
+      backfill({
+        source: 'SDEX',
+        pair: 'XLM/USDC',
+        start: new Date('2026-10-16T10:00:00Z'),
+        end: new Date('2026-10-16T08:00:00Z'),
+      })
+    ).rejects.toThrow(/start .* must be before end/)
+
+    await expect(
+      backfill({
+        source: 'SDEX',
+        pair: 'XLM/USDC',
+        start: new Date('2026-10-16T00:00:00Z'),
+        end: new Date(Date.now() + 86400_000 * 365), // 1 year in future
+      })
+    ).rejects.toThrow(/cannot be in the future/)
+  })
+
+  it('performs a successful backfill that fills an open gap', async () => {
+    const start = new Date('2026-10-15T00:00:00Z')
+    const end = new Date('2026-10-15T02:00:00Z')
+
+    // Seed open gap
+    await recordGaps({
+      network: 'testnet',
+      pair: 'XLM/USDC',
+      source: 'SDEX',
+      gaps: [{ start, end, status: 'open' }],
+    })
+    expect(storedGaps[0].status).toBe('open')
+
+    // Mock Horizon returning trades with valid TOID token (e.g. ledger 50000)
+    mockHorizonTrades.mockResolvedValueOnce({
+      records: [
+        {
+          id: 'trade-1',
+          paging_token: `${(50000n << 32n).toString()}-1`,
+          ledger_close_time: '2026-10-15T01:00:00Z',
+          base_asset_type: 'native',
+          counter_asset_code: 'USDC',
+          price: { n: 25, d: 100 },
+          base_amount: '100',
+          counter_amount: '25',
+        },
+      ],
+    }).mockResolvedValueOnce({ records: [] })
+
+    const result = await backfill({
+      source: 'SDEX',
+      pair: 'XLM/USDC',
+      start,
+      end,
+      network: 'testnet',
+    })
+
+    expect(result.success).toBe(true)
+    expect(result.pointsIngested).toBe(1)
+    expect(result.gapsFilled).toBe(1)
+
+    // Verify stored gap is updated to filled
+    expect(storedGaps[0].status).toBe('filled')
+  })
+
+  it('proves idempotency on re-run: second run inserts 0 points and changes nothing', async () => {
+    const start = new Date('2026-10-15T00:00:00Z')
+    const end = new Date('2026-10-15T02:00:00Z')
+
+    await recordGaps({
+      network: 'testnet',
+      pair: 'XLM/USDC',
+      source: 'SDEX',
+      gaps: [{ start, end, status: 'open' }],
+    })
+
+    const sampleRecords = [
+      {
+        id: 'trade-repeat',
+        paging_token: `${(50000n << 32n).toString()}-1`,
+        ledger_close_time: '2026-10-15T01:00:00Z',
+        base_asset_type: 'native',
+        counter_asset_code: 'USDC',
+        price: { n: 25, d: 100 },
+        base_amount: '100',
+        counter_amount: '25',
+      },
+    ]
+
+    mockHorizonTrades
+      .mockResolvedValueOnce({ records: sampleRecords })
+      .mockResolvedValueOnce({ records: [] })
+
+    // Run 1: fills gap
+    const run1 = await backfill({
+      source: 'SDEX',
+      pair: 'XLM/USDC',
+      start,
+      end,
+      network: 'testnet',
+    })
+    expect(run1.pointsIngested).toBe(1)
+    expect(storedPoints).toHaveLength(1)
+    expect(storedGaps[0].status).toBe('filled')
+
+    // Run 2: exact same range & trades
+    mockHorizonTrades
+      .mockResolvedValueOnce({ records: sampleRecords })
+      .mockResolvedValueOnce({ records: [] })
+
+    const run2 = await backfill({
+      source: 'SDEX',
+      pair: 'XLM/USDC',
+      start,
+      end,
+      network: 'testnet',
+    })
+
+    // Idempotent: 0 new points inserted, no duplicate points in storage
+    expect(run2.pointsIngested).toBe(0)
+    expect(storedPoints).toHaveLength(1)
+    expect(storedGaps).toHaveLength(1)
+    expect(storedGaps[0].status).toBe('filled')
+  })
+})
+
+describe('Permanent Gaps (Non-history Sources)', () => {
+  beforeEach(() => {
+    storedGaps.length = 0
+    storedPoints.length = 0
+    vi.clearAllMocks()
+    ;(globalThis as any).__lens_prisma = mockPrisma
+    ;(globalThis as any).__lens_pgPool = mockPgPool
+  })
+
+  it('marks newly detected gaps for non-history sources as permanent', async () => {
+    const start = new Date('2026-10-15T00:00:00Z')
+    const end = new Date('2026-10-15T04:00:00Z')
+
+    const detected = await detectGaps({
+      network: 'testnet',
+      pair: 'XLM/USDC',
+      source: 'soroswap_amm',
+      start,
+      end,
+      cadenceMs: 300_000,
+    })
+
+    expect(detected).toHaveLength(1)
+    expect(detected[0].source).toBe('soroswap_amm')
+    expect(detected[0].status).toBe('permanent')
+    expect(storedGaps[0].status).toBe('permanent')
+  })
+
+  it('reports clearly that an unfillable gap cannot be backfilled and marks it permanent', async () => {
+    const start = new Date('2026-10-15T00:00:00Z')
+    const end = new Date('2026-10-15T02:00:00Z')
+
+    const result = await backfill({
+      source: 'aquarius_amm',
+      pair: 'XLM/USDC',
+      start,
+      end,
+      network: 'testnet',
+    })
+
+    expect(result.success).toBe(false)
+    expect(result.unfillable).toBe(true)
+    expect(result.message).toMatch(/does not support historical data fetch.*marked permanent/)
+    expect(storedGaps).toHaveLength(1)
+    expect(storedGaps[0].status).toBe('permanent')
+  })
+})
+
+describe('Incomplete Window Declarations in API Responses', () => {
+  beforeEach(() => {
+    storedGaps.length = 0
+    storedPoints.length = 0
+    vi.clearAllMocks()
+    ;(globalThis as any).__lens_prisma = mockPrisma
+    ;(globalThis as any).__lens_pgPool = mockPgPool
+  })
+
+  async function buildTestApp() {
+    const app = Fastify({ logger: false })
+    await app.register(registerNetworkSelector)
+    await registerPriceRoutes(app)
+    await registerHistoryRoutes(app)
+    await registerCandleRoutes(app)
+    await registerRESTRoutes(app)
+    await app.ready()
+    return app
+  }
+
+  it('flags response as incomplete with overlapping gap details when an open gap overlaps the window', async () => {
+    const gapStart = new Date('2026-10-15T01:00:00Z')
+    const gapEnd = new Date('2026-10-15T03:00:00Z')
+
+    // Record open gap
+    await recordGaps({
+      network: 'testnet',
+      pair: 'XLM/USDC',
+      source: 'SDEX',
+      gaps: [{ start: gapStart, end: gapEnd, status: 'open' }],
+    })
+
+    const app = await buildTestApp()
+
+    // Request window overlapping the gap: 00:00 → 04:00
+    const res = await app.inject({
+      method: 'GET',
+      url: '/prices/history?pair=XLM/USDC&from=2026-10-15T00:00:00.000Z&to=2026-10-15T04:00:00.000Z',
+    })
+
+    expect(res.statusCode).toBe(200)
+    const body = res.json()
+    expect(body.incomplete).toBe(true)
+    expect(body.gaps).toBeDefined()
+    expect(body.gaps).toHaveLength(1)
+    expect(body.gaps[0].source).toBe('SDEX')
+    expect(body.gaps[0].status).toBe('open')
+    expect(body.gaps[0].start).toBe(gapStart.toISOString())
+    expect(body.gaps[0].end).toBe(gapEnd.toISOString())
+  })
+
+  it('flags response as incomplete when a permanent gap overlaps the window', async () => {
+    const gapStart = new Date('2026-10-15T01:00:00Z')
+    const gapEnd = new Date('2026-10-15T02:00:00Z')
+
+    await recordGaps({
+      network: 'testnet',
+      pair: 'XLM/USDC',
+      source: 'soroswap_amm',
+      gaps: [{ start: gapStart, end: gapEnd, status: 'permanent' }],
+    })
+
+    const app = await buildTestApp()
+
+    const res = await app.inject({
+      method: 'GET',
+      url: '/candles/XLM/USDC?from=2026-10-15T00:00:00.000Z&to=2026-10-15T03:00:00.000Z',
+    })
+
+    expect(res.statusCode).toBe(200)
+    const body = res.json()
+    expect(body.incomplete).toBe(true)
+    expect(body.gaps[0].source).toBe('soroswap_amm')
+    expect(body.gaps[0].status).toBe('permanent')
+  })
+
+  it('does NOT flag response when the window covers only filled gaps or no gaps', async () => {
+    // Record a FILLED gap in that window
+    await recordGaps({
+      network: 'testnet',
+      pair: 'XLM/USDC',
+      source: 'SDEX',
+      gaps: [{
+        start: new Date('2026-10-15T01:00:00Z'),
+        end: new Date('2026-10-15T02:00:00Z'),
+        status: 'filled',
+      }],
+    })
+
+    const app = await buildTestApp()
+
+    // Request window
+    const res = await app.inject({
+      method: 'GET',
+      url: '/prices/history?pair=XLM/USDC&from=2026-10-15T00:00:00.000Z&to=2026-10-15T03:00:00.000Z',
+    })
+
+    expect(res.statusCode).toBe(200)
+    const body = res.json()
+    // Should NOT be flagged incomplete
+    expect(body.incomplete).toBeUndefined()
+    expect(body.gaps).toBeUndefined()
+  })
+})
diff --git a/src/api/history.ts b/src/api/history.ts
index df8df80d..1d26d994 100644
--- a/src/api/history.ts
+++ b/src/api/history.ts
@@ -2,6 +2,7 @@ import type { FastifyInstance } from 'fastify'
 import { pgPool } from '../db'
 import { activeNetwork, type NetworkName } from '../config'
 import '../middleware/network' // declares req.network on the FastifyRequest type
+import { findOverlappingGaps } from '../backfill/gaps'
 
 /** Supported aggregation intervals → bucket width in seconds. */
 export const HISTORY_INTERVAL_SECONDS: Record<string, number> = {
@@ -111,6 +112,14 @@ export async function registerHistoryRoutes(app: FastifyInstance) {
 
     const network = req.network ?? activeNetwork
     const points = await queryHistory(pair, from, to, intervalSecs, network)
+    const overlappingGaps = await findOverlappingGaps({
+      network,
+      pair,
+      start: from,
+      end: to,
+    })
+
+    const gapDeclaration = overlappingGaps.length > 0 ? { incomplete: true, gaps: overlappingGaps } : {}
 
     return {
       pair,
@@ -120,6 +129,7 @@ export async function registerHistoryRoutes(app: FastifyInstance) {
       to: to.toISOString(),
       count: points.length,
       points,
+      ...gapDeclaration,
     }
   })
 }
diff --git a/src/api/rest.ts b/src/api/rest.ts
index 8f386e6a..58d0e2d6 100644
--- a/src/api/rest.ts
+++ b/src/api/rest.ts
@@ -16,6 +16,7 @@ import {
   installResponseValidation,
 } from './schemas'
 import { getDepth } from '../pricing/depth'
+import { findOverlappingGaps } from '../backfill/gaps'
 
 function makePairKey(a: string, b: string): string {
   return [a, b].sort().join('/')
@@ -164,6 +165,7 @@ export async function registerRESTRoutes(app: FastifyInstance) {
     { schema: { response: { 200: historyResponseSchema } } },
     async (req, reply) => {
       const { assetA, assetB } = req.params
+      const network = req.network ?? activeNetwork
       const window = req.query.window ?? '1h'
       const limit = Math.min(parseInt(req.query.limit ?? '100', 10), 1000)
       const pairKey = makePairKey(assetA, assetB)
@@ -183,6 +185,25 @@ export async function registerRESTRoutes(app: FastifyInstance) {
         [pairKey, window, limit]
       )
 
+      const windowMs: Record<string, number> = {
+        '1m': 60_000,
+        '5m': 300_000,
+        '1h': 3600_000,
+        '24h': 86400_000,
+      }
+      const span = (windowMs[window] ?? 3600_000) * limit
+      const now = new Date()
+      const queryStart = new Date(now.getTime() - span)
+
+      const overlappingGaps = await findOverlappingGaps({
+        network,
+        pair: pairKey,
+        start: queryStart,
+        end: now,
+      })
+
+      const gapDeclaration = overlappingGaps.length > 0 ? { incomplete: true, gaps: overlappingGaps } : {}
+
       return {
         pairKey,
         window,
@@ -198,6 +219,7 @@ export async function registerRESTRoutes(app: FastifyInstance) {
           high: r.high_price,
           low: r.low_price,
         })),
+        ...gapDeclaration,
       }
     }
   )
diff --git a/src/api/schemas.ts b/src/api/schemas.ts
index 90ecd10f..db14a05e 100644
--- a/src/api/schemas.ts
+++ b/src/api/schemas.ts
@@ -118,6 +118,21 @@ export const historyResponseSchema = {
   properties: {
     pairKey: { type: 'string' },
     window: { type: 'string', enum: ['1m', '5m', '1h', '24h'] },
+    incomplete: { type: 'boolean' },
+    gaps: {
+      type: 'array',
+      items: {
+        type: 'object',
+        required: ['source', 'start', 'end', 'status'],
+        additionalProperties: false,
+        properties: {
+          source: { type: 'string' },
+          start: { type: 'string' },
+          end: { type: 'string' },
+          status: { type: 'string', enum: ['open', 'permanent'] },
+        },
+      },
+    },
     buckets: {
       type: 'array',
       items: {
diff --git a/src/backfill/backfillService.ts b/src/backfill/backfillService.ts
new file mode 100644
index 00000000..937e35f4
--- /dev/null
+++ b/src/backfill/backfillService.ts
@@ -0,0 +1,302 @@
+import { Asset } from '@stellar/stellar-sdk'
+import { activeNetwork, getNetworkConfig, type NetworkName } from '../config'
+import { getHorizonServer } from '../network/clients'
+import { upsertPricePoints } from '../db'
+import { resolvePageLedgers } from '../ingesters/toid'
+import { deriveSdexPricePoints, type SdexRawTrades } from '../ingest/venues/sdex'
+import { getActivePairs, parseAssetStr, makePairKey } from '../pairsRegistry'
+import type { WatchedPair, AssetId } from '../types'
+import type { VenuePricePoint } from '../ingest/venues/types'
+import {
+  supportsHistory,
+  markGapsFilled,
+  markGapsPermanent,
+} from './gaps'
+import type { BackfillResult } from './types'
+
+function toAsset(asset: { code: string; issuer: string | null }): Asset {
+  if (!asset.issuer || asset.code === 'XLM') return Asset.native()
+  return new Asset(asset.code, asset.issuer)
+}
+
+function resolvePair(pairStr: string, network: NetworkName): WatchedPair {
+  const active = getNetworkConfig(network).pairs.find(
+    p => p.pairKey.toUpperCase() === pairStr.toUpperCase()
+  ) ?? getActivePairs().find(
+    p => p.pairKey.toUpperCase() === pairStr.toUpperCase()
+  )
+
+  if (active) return active
+
+  // Try parsing "CODE:ISSUER/CODE:ISSUER" or "CODE/CODE"
+  const [aStr, bStr] = pairStr.split('/')
+  if (!aStr || !bStr) {
+    throw new Error(`Invalid pair format: "${pairStr}". Expected "ASSET_A/ASSET_B"`)
+  }
+
+  const assetA = parseAssetStr(aStr) ?? { code: aStr.toUpperCase(), issuer: null }
+  const assetB = parseAssetStr(bStr) ?? { code: bStr.toUpperCase(), issuer: null }
+  return {
+    assetA,
+    assetB,
+    pairKey: makePairKey(assetA, assetB),
+  }
+}
+
+export interface BackfillOptions {
+  source: string
+  pair: string
+  start: Date
+  end: Date
+  network?: NetworkName
+}
+
+/**
+ * Executes a backfill for a given source, pair, and time range.
+ *
+ * 1. Validates input range (start < end, not in the future).
+ * 2. Checks if the source supports historical data fetch.
+ *    If NOT: marks any overlapping gaps as 'permanent' and reports unfillable without failing noisily.
+ * 3. If supported: refetches data from the source and writes it through the SAME
+ *    idempotent write path as live ingestion (`upsertPricePoints`).
+ * 4. Running the same backfill twice produces no duplicates and no changes the second time.
+ * 5. Updates gap records as ranges get filled.
+ */
+export async function backfill(options: BackfillOptions): Promise<BackfillResult> {
+  const network = options.network ?? activeNetwork
+  const { source, start, end } = options
+
+  // Validate range
+  if (isNaN(start.getTime())) {
+    throw new Error('Invalid start date: must be a valid ISO 8601 date string')
+  }
+  if (isNaN(end.getTime())) {
+    throw new Error('Invalid end date: must be a valid ISO 8601 date string')
+  }
+  if (start >= end) {
+    throw new Error(`Invalid time range: start (${start.toISOString()}) must be before end (${end.toISOString()})`)
+  }
+  const now = new Date()
+  if (end.getTime() > now.getTime() + 60_000) {
+    throw new Error(`Invalid time range: end (${end.toISOString()}) cannot be in the future`)
+  }
+
+  const pair = resolvePair(options.pair, network)
+
+  // 1. Check history capability
+  if (!supportsHistory(source)) {
+    await markGapsPermanent({
+      network,
+      pair: pair.pairKey,
+      source,
+      start,
+      end,
+    })
+
+    const message = `Source "${source}" does not support historical data fetch; gap [${start.toISOString()} - ${end.toISOString()}] cannot be filled and has been marked permanent.`
+    console.warn(`[backfill] ${message}`)
+
+    return {
+      success: false,
+      unfillable: true,
+      source,
+      pair: pair.pairKey,
+      network,
+      start,
+      end,
+      pointsIngested: 0,
+      gapsFilled: 0,
+      message,
+    }
+  }
+
+  // 2. Fetch history for supported source
+  let pointsIngested = 0
+
+  if (source.toUpperCase() === 'SDEX') {
+    pointsIngested = await backfillSdexTrades(pair, start, end, network)
+  } else if (source.toUpperCase() === 'AMM') {
+    pointsIngested = await backfillAmmTrades(pair, start, end, network)
+  }
+
+  // 3. Update gap records (fully or partially covered)
+  const gapsFilled = await markGapsFilled({
+    network,
+    pair: pair.pairKey,
+    source,
+    start,
+    end,
+  })
+
+  return {
+    success: true,
+    source,
+    pair: pair.pairKey,
+    network,
+    start,
+    end,
+    pointsIngested,
+    gapsFilled,
+  }
+}
+
+/**
+ * Backfills SDEX trades by paging Horizon across [start, end].
+ * Writes derived points directly through `upsertPricePoints`.
+ */
+async function backfillSdexTrades(
+  pair: WatchedPair,
+  start: Date,
+  end: Date,
+  network: NetworkName
+): Promise<number> {
+  const assetA = toAsset(pair.assetA)
+  const assetB = toAsset(pair.assetB)
+  const horizon = getHorizonServer(network)
+
+  let totalIngested = 0
+  let cursor = '0'
+  let hasMore = true
+
+  // Loop pages of trades until reaching trades past `end`
+  while (hasMore) {
+    try {
+      const page = await horizon
+        .trades()
+        .forAssetPair(assetA, assetB)
+        .cursor(cursor)
+        .limit(200)
+        .order('asc')
+        .call()
+
+      const records = page.records
+      if (!records || records.length === 0) {
+        break
+      }
+
+      // Filter records within [start, end]
+      const inRangeRecords = records.filter((t: any) => {
+        const time = new Date(t.ledger_close_time).getTime()
+        return time >= start.getTime() && time <= end.getTime()
+      })
+
+      if (inRangeRecords.length > 0) {
+        const raw: SdexRawTrades = {
+          records: inRangeRecords,
+          cursor,
+          stateId: `sdex:backfill:${pair.pairKey}`,
+          lastLedger: null,
+        }
+
+        const derived = deriveSdexPricePoints(raw, pair)
+        if (derived && derived.points.length > 0) {
+          const count = await upsertPricePoints(derived.points, network)
+          totalIngested += count
+        }
+      }
+
+      const lastRecord = records[records.length - 1]
+      cursor = lastRecord.paging_token
+
+      // If the last trade in this page is past `end`, we are done
+      if (new Date(lastRecord.ledger_close_time).getTime() > end.getTime()) {
+        hasMore = false
+      }
+    } catch (err) {
+      console.error(`[backfill] Error paging SDEX trades for ${pair.pairKey}:`, (err as Error).message)
+      break
+    }
+  }
+
+  return totalIngested
+}
+
+/**
+ * Backfills AMM liquidity pool trades by paging Horizon across [start, end].
+ * Writes derived points directly through `upsertPricePoints`.
+ */
+async function backfillAmmTrades(
+  pair: WatchedPair,
+  start: Date,
+  end: Date,
+  network: NetworkName
+): Promise<number> {
+  let totalIngested = 0
+  const netConfig = getNetworkConfig(network)
+
+  try {
+    const assetAStr = pair.assetA.issuer ? `${pair.assetA.code}:${pair.assetA.issuer}` : 'native'
+    const assetBStr = pair.assetB.issuer ? `${pair.assetB.code}:${pair.assetB.issuer}` : 'native'
+
+    const params = new URLSearchParams()
+    params.append('reserves[]', assetAStr)
+    params.append('reserves[]', assetBStr)
+
+    const response = await fetch(`${netConfig.horizon.url}/liquidity_pools?${params.toString()}`)
+    const data = (await response.json()) as any
+    const pools = data._embedded?.records ?? []
+
+    for (const pool of pools) {
+      let cursor = '0'
+      let hasMore = true
+
+      while (hasMore) {
+        const tradesRes = await fetch(
+          `${netConfig.horizon.url}/liquidity_pools/${pool.id}/trades?cursor=${cursor}&limit=200&order=asc`
+        )
+        const tradesData = (await tradesRes.json()) as any
+        const records = tradesData._embedded?.records ?? []
+        if (records.length === 0) break
+
+        const inRange = records.filter((t: any) => {
+          const time = new Date(t.ledger_close_time).getTime()
+          return time >= start.getTime() && time <= end.getTime()
+        })
+
+        if (inRange.length > 0) {
+          const ledgers = resolvePageLedgers(
+            inRange.map((t: any) => t.paging_token ?? t.id),
+            null
+          )
+
+          if (ledgers !== null) {
+            const points: VenuePricePoint[] = inRange.map((t: any, i: number) => {
+              const baseCode = t.base_asset_type === 'native' ? 'XLM' : t.base_asset_code
+              const isForward = baseCode === pair.assetA.code
+              const price = isForward
+                ? parseFloat(t.counter_amount) / parseFloat(t.base_amount)
+                : parseFloat(t.base_amount) / parseFloat(t.counter_amount)
+
+              return {
+                assetA: pair.assetA.code,
+                assetB: pair.assetB.code,
+                pairKey: pair.pairKey,
+                source: 'AMM',
+                poolId: pool.id,
+                price,
+                baseVolume: parseFloat(t.base_amount),
+                counterVolume: parseFloat(t.counter_amount),
+                ledger: ledgers[i],
+                timestamp: new Date(t.ledger_close_time),
+                eventId: t.id,
+              }
+            })
+
+            const count = await upsertPricePoints(points, network)
+            totalIngested += count
+          }
+        }
+
+        const lastRec = records[records.length - 1]
+        cursor = lastRec.paging_token
+        if (new Date(lastRec.ledger_close_time).getTime() > end.getTime()) {
+          hasMore = false
+        }
+      }
+    }
+  } catch (err) {
+    console.error(`[backfill] Error paging AMM trades for ${pair.pairKey}:`, (err as Error).message)
+  }
+
+  return totalIngested
+}
diff --git a/src/backfill/gaps.ts b/src/backfill/gaps.ts
new file mode 100644
index 00000000..0282b604
--- /dev/null
+++ b/src/backfill/gaps.ts
@@ -0,0 +1,443 @@
+import type { NetworkName } from '../config'
+
+function getPrisma() {
+  return (globalThis as any).__lens_prisma
+}
+
+function getPgPool() {
+  return (globalThis as any).__lens_pgPool
+}
+import type { GapRecord, GapStatus, OverlappingGapSummary } from './types'
+
+interface GapRow {
+  id: string
+  network: string
+  pair: string
+  source: string
+  start: Date
+  end: Date
+  status: string
+  detectedAt?: Date
+  updatedAt?: Date
+}
+
+/** Expected maximum cadence between data points before an interval is considered a gap. */
+export const EXPECTED_CADENCE_MS: Record<string, number> = {
+  SDEX: 300_000, // 5 minutes
+  AMM: 300_000, // 5 minutes
+  soroswap_amm: 300_000, // 5 minutes (poll is 60s; 5 missed cycles = hole)
+  aquarius_amm: 300_000, // 5 minutes
+  default: 300_000,
+}
+
+/** Returns the expected cadence in milliseconds for a source. */
+export function getExpectedCadenceMs(source: string): number {
+  return EXPECTED_CADENCE_MS[source] ?? EXPECTED_CADENCE_MS.default
+}
+
+/**
+ * Determines whether a source supports historical backfill.
+ * SDEX and AMM trades can be paged from Stellar Horizon.
+ * Soroswap and Aquarius are state-only snapshot APIs without historical trade endpoints.
+ */
+export function supportsHistory(source: string): boolean {
+  const norm = source.toUpperCase()
+  return norm === 'SDEX' || norm === 'AMM'
+}
+
+export interface DetectGapsOptions {
+  network: NetworkName
+  pair: string
+  source: string
+  start?: Date
+  end?: Date
+  cadenceMs?: number
+}
+
+/**
+ * Detects holes in ingested price points for a specific (source, pair, network).
+ * A hole is any period >= cadenceMs without price points.
+ * Records the detected gaps in persistent storage idempotently.
+ */
+export async function detectGaps(options: DetectGapsOptions): Promise<GapRecord[]> {
+  const pgPool = getPgPool()
+  if (!pgPool?.query) return []
+  const { network, pair, source } = options
+  const cadenceMs = options.cadenceMs ?? getExpectedCadenceMs(source)
+  const end = options.end ?? new Date()
+
+  // Determine scan start: use provided start, or find earliest point, or default to 24h lookback
+  let start = options.start
+  if (!start) {
+    const earliestResult = await pgPool.query(
+      `SELECT MIN(timestamp) as earliest FROM price_points
+       WHERE network = $1 AND pair_key = $2 AND source = $3`,
+      [network, pair, source]
+    )
+    const earliest = earliestResult.rows[0]?.earliest
+    start = earliest ? new Date(earliest) : new Date(end.getTime() - 24 * 60 * 60 * 1000)
+  }
+
+  if (start >= end) {
+    return []
+  }
+
+  // Fetch all existing price points in [start, end] ordered by timestamp
+  const pointsResult = await pgPool.query(
+    `SELECT timestamp FROM price_points
+     WHERE network = $1 AND pair_key = $2 AND source = $3
+       AND timestamp >= $4 AND timestamp <= $5
+     ORDER BY timestamp ASC`,
+    [network, pair, source, start, end]
+  )
+
+  const timestamps = pointsResult.rows.map((r: any) => new Date(r.timestamp))
+  const rawGaps: Array<{ start: Date; end: Date }> = []
+
+  if (timestamps.length === 0) {
+    // Entire window is missing
+    if (end.getTime() - start.getTime() >= cadenceMs) {
+      rawGaps.push({ start, end })
+    }
+  } else {
+    // Check leading edge (start -> first point)
+    if (timestamps[0].getTime() - start.getTime() >= cadenceMs) {
+      rawGaps.push({ start, end: timestamps[0] })
+    }
+
+    // Check interior gaps between successive points
+    for (let i = 0; i < timestamps.length - 1; i++) {
+      const diff = timestamps[i + 1].getTime() - timestamps[i].getTime()
+      if (diff >= cadenceMs) {
+        rawGaps.push({ start: timestamps[i], end: timestamps[i + 1] })
+      }
+    }
+
+    // Check trailing edge (last point -> end)
+    const lastPoint = timestamps[timestamps.length - 1]
+    if (end.getTime() - lastPoint.getTime() >= cadenceMs) {
+      rawGaps.push({ start: lastPoint, end })
+    }
+  }
+
+  if (rawGaps.length === 0) {
+    return []
+  }
+
+  return await recordGaps({
+    network,
+    pair,
+    source,
+    gaps: rawGaps,
+  })
+}
+
+export interface RecordGapsInput {
+  network: NetworkName
+  pair: string
+  source: string
+  gaps: Array<{ start: Date; end: Date; status?: GapStatus }>
+}
+
+/**
+ * Idempotently records gap intervals into `ingestion_gaps`.
+ * Merges adjacent or overlapping gaps with existing persistent rows.
+ * If the source cannot provide history, gaps are automatically marked 'permanent'.
+ */
+export async function recordGaps(input: RecordGapsInput): Promise<GapRecord[]> {
+  const prisma = getPrisma()
+  if (!prisma?.ingestionGap?.findMany) return []
+  const { network, pair, source, gaps } = input
+  if (gaps.length === 0) return []
+
+  const sourceSupportsHistory = supportsHistory(source)
+  const defaultStatus: GapStatus = sourceSupportsHistory ? 'open' : 'permanent'
+
+  const savedGaps: GapRecord[] = []
+
+  for (const gap of gaps) {
+    const desiredStatus = !sourceSupportsHistory ? 'permanent' : (gap.status ?? defaultStatus)
+
+    // Find any existing gap rows for this (network, pair, source) that overlap or touch [gap.start, gap.end]
+    // Tolerance for adjacency: within 1 second
+    const overlapping: GapRow[] = await prisma.ingestionGap.findMany({
+      where: {
+        network,
+        pair,
+        source,
+        AND: [
+          { start: { lte: new Date(gap.end.getTime() + 1000) } },
+          { end: { gte: new Date(gap.start.getTime() - 1000) } },
+        ],
+      },
+      orderBy: { start: 'asc' },
+    })
+
+    if (overlapping.length === 0) {
+      // Clean insert
+      const created = await prisma.ingestionGap.create({
+        data: {
+          network,
+          pair,
+          source,
+          start: gap.start,
+          end: gap.end,
+          status: desiredStatus,
+        },
+      })
+      savedGaps.push({
+        id: created.id,
+        network: created.network as NetworkName,
+        pair: created.pair,
+        source: created.source,
+        start: created.start,
+        end: created.end,
+        status: created.status as GapStatus,
+        detectedAt: created.detectedAt,
+        updatedAt: created.updatedAt,
+      })
+    } else {
+      // Merge with existing overlapping/adjacent rows
+      const mergedStart = new Date(Math.min(gap.start.getTime(), ...overlapping.map(o => o.start.getTime())))
+      const mergedEnd = new Date(Math.max(gap.end.getTime(), ...overlapping.map(o => o.end.getTime())))
+
+      // Determine merged status:
+      // - If source has no history capability -> permanent
+      // - If any existing was permanent -> permanent
+      // - If all existing were filled and new gap is completely inside filled range -> filled
+      // - Otherwise -> open
+      let mergedStatus: GapStatus = desiredStatus
+      if (!sourceSupportsHistory || overlapping.some(o => o.status === 'permanent')) {
+        mergedStatus = 'permanent'
+      } else if (overlapping.every(o => o.status === 'filled') && gap.status === 'filled') {
+        mergedStatus = 'filled'
+      } else if (overlapping.some(o => o.status === 'open') || desiredStatus === 'open') {
+        mergedStatus = 'open'
+      }
+
+      const primary = overlapping[0]
+      const updated = await prisma.ingestionGap.update({
+        where: { id: primary.id },
+        data: {
+          start: mergedStart,
+          end: mergedEnd,
+          status: mergedStatus,
+          updatedAt: new Date(),
+        },
+      })
+
+      // Delete extra redundant overlapping rows that were consolidated
+      if (overlapping.length > 1) {
+        const extraIds = overlapping.slice(1).map(o => o.id)
+        await prisma.ingestionGap.deleteMany({
+          where: { id: { in: extraIds } },
+        })
+      }
+
+      savedGaps.push({
+        id: updated.id,
+        network: updated.network as NetworkName,
+        pair: updated.pair,
+        source: updated.source,
+        start: updated.start,
+        end: updated.end,
+        status: updated.status as GapStatus,
+        detectedAt: updated.detectedAt,
+        updatedAt: updated.updatedAt,
+      })
+    }
+  }
+
+  return savedGaps
+}
+
+function getPairVariants(pair: string): string[] {
+  if (pair.includes('/')) {
+    const [a, b] = pair.split('/')
+    return a === b ? [pair] : [pair, `${b}/${a}`]
+  }
+  return [pair]
+}
+
+export interface FindOverlappingGapsOptions {
+  network: NetworkName
+  pair: string
+  start: Date
+  end: Date
+  source?: string
+}
+
+/**
+ * Finds all known gaps (status 'open' or 'permanent') that overlap a given time window.
+ * Returns an array of gap summaries. Filled gaps are excluded since the window is complete.
+ */
+export async function findOverlappingGaps(options: FindOverlappingGapsOptions): Promise<OverlappingGapSummary[]> {
+  const prisma = getPrisma()
+  if (!prisma?.ingestionGap?.findMany) {
+    return []
+  }
+  const { network, pair, start, end, source } = options
+  const pairVariants = getPairVariants(pair)
+
+  const whereClause: any = {
+    network,
+    pair: pairVariants.length > 1 ? { in: pairVariants } : pair,
+    status: { in: ['open', 'permanent'] },
+    start: { lt: end },
+    end: { gt: start },
+  }
+
+  if (source) {
+    whereClause.source = source
+  }
+
+  const gaps: GapRow[] = await prisma.ingestionGap.findMany({
+    where: whereClause,
+    orderBy: { start: 'asc' },
+  })
+
+  return gaps.map((g: any) => ({
+    source: g.source,
+    start: g.start.toISOString(),
+    end: g.end.toISOString(),
+    status: g.status as 'open' | 'permanent',
+  }))
+}
+
+export interface MarkGapsFilledOptions {
+  network: NetworkName
+  pair: string
+  source: string
+  start: Date
+  end: Date
+}
+
+/**
+ * Updates gap records as a range gets filled (partially or fully).
+ * A gap covered completely is marked 'filled'.
+ * A gap covered partially is split into filled and open portions.
+ */
+export async function markGapsFilled(options: MarkGapsFilledOptions): Promise<number> {
+  const prisma = getPrisma()
+  if (!prisma?.ingestionGap?.findMany) return 0
+  const { network, pair, source, start, end } = options
+
+  const pairVariants = getPairVariants(pair)
+  // Find all open gaps overlapping the filled range
+  const overlapping: GapRow[] = await prisma.ingestionGap.findMany({
+    where: {
+      network,
+      pair: pairVariants.length > 1 ? { in: pairVariants } : pair,
+      source,
+      status: 'open',
+      start: { lt: end },
+      end: { gt: start },
+    },
+    orderBy: { start: 'asc' },
+  })
+
+  let modifiedCount = 0
+
+  for (const gap of overlapping) {
+    const isCoveredCompletely = start <= gap.start && end >= gap.end
+    if (isCoveredCompletely) {
+      await prisma.ingestionGap.update({
+        where: { id: gap.id },
+        data: { status: 'filled', updatedAt: new Date() },
+      })
+      modifiedCount++
+    } else if (start <= gap.start && end < gap.end) {
+      // Partial fill on the left side
+      // 1. Update existing row to represent the filled range [gap.start, end]
+      await prisma.ingestionGap.update({
+        where: { id: gap.id },
+        data: { end, status: 'filled', updatedAt: new Date() },
+      })
+      // 2. Insert new row for remaining open range [end, gap.end]
+      await prisma.ingestionGap.create({
+        data: {
+          network,
+          pair,
+          source,
+          start: end,
+          end: gap.end,
+          status: 'open',
+        },
+      })
+      modifiedCount++
+    } else if (start > gap.start && end >= gap.end) {
+      // Partial fill on the right side
+      // 1. Update existing row to represent the remaining open range [gap.start, start]
+      await prisma.ingestionGap.update({
+        where: { id: gap.id },
+        data: { end: start, status: 'open', updatedAt: new Date() },
+      })
+      // 2. Insert new row for filled range [start, gap.end]
+      await prisma.ingestionGap.create({
+        data: {
+          network,
+          pair,
+          source,
+          start,
+          end: gap.end,
+          status: 'filled',
+        },
+      })
+      modifiedCount++
+    } else {
+      // Middle fill: gap.start < start && end < gap.end
+      // 1. Left remaining open: [gap.start, start]
+      await prisma.ingestionGap.update({
+        where: { id: gap.id },
+        data: { end: start, status: 'open', updatedAt: new Date() },
+      })
+      // 2. Middle filled: [start, end]
+      await prisma.ingestionGap.create({
+        data: {
+          network,
+          pair,
+          source,
+          start,
+          end,
+          status: 'filled',
+        },
+      })
+      // 3. Right remaining open: [end, gap.end]
+      await prisma.ingestionGap.create({
+        data: {
+          network,
+          pair,
+          source,
+          start: end,
+          end: gap.end,
+          status: 'open',
+        },
+      })
+      modifiedCount++
+    }
+  }
+
+  return modifiedCount
+}
+
+export interface MarkGapsPermanentOptions {
+  network: NetworkName
+  pair: string
+  source: string
+  start: Date
+  end: Date
+}
+
+/**
+ * Marks overlapping gaps permanent, or creates a permanent gap if none exists.
+ * Used when attempting backfill against an unfillable source.
+ */
+export async function markGapsPermanent(options: MarkGapsPermanentOptions): Promise<GapRecord[]> {
+  const { network, pair, source, start, end } = options
+  return await recordGaps({
+    network,
+    pair,
+    source,
+    gaps: [{ start, end, status: 'permanent' }],
+  })
+}
diff --git a/src/backfill/types.ts b/src/backfill/types.ts
new file mode 100644
index 00000000..e0c18224
--- /dev/null
+++ b/src/backfill/types.ts
@@ -0,0 +1,40 @@
+import type { NetworkName } from '../config'
+
+export type GapStatus = 'open' | 'filled' | 'permanent'
+
+export interface GapRecord {
+  id: string
+  network: NetworkName
+  source: string
+  pair: string
+  start: Date
+  end: Date
+  status: GapStatus
+  detectedAt: Date
+  updatedAt: Date
+}
+
+export interface OverlappingGapSummary {
+  source: string
+  start: string
+  end: string
+  status: 'open' | 'permanent'
+}
+
+export interface IncompleteWindowDeclaration {
+  incomplete?: boolean
+  gaps?: OverlappingGapSummary[]
+}
+
+export interface BackfillResult {
+  success: boolean
+  unfillable?: boolean
+  source: string
+  pair: string
+  network: NetworkName
+  start: Date
+  end: Date
+  pointsIngested: number
+  gapsFilled: number
+  message?: string
+}
diff --git a/src/db.ts b/src/db.ts
index a98b66cc..8a01de73 100644
--- a/src/db.ts
+++ b/src/db.ts
@@ -9,6 +9,7 @@ export const prisma = globalForPrisma.prisma ?? new PrismaClient({
   log: process.env.NODE_ENV === 'development' ? ['warn', 'error'] : ['error'],
 })
 if (process.env.NODE_ENV !== 'production') globalForPrisma.prisma = prisma
+;(globalThis as any).__lens_prisma = prisma
 
 // Raw pg pool for time-series queries (VWAP, aggregates)
 // ssl: true with rejectUnauthorized: false is required for Supabase pgbouncer
@@ -17,6 +18,7 @@ export const pgPool = new Pool({
   connectionString: config.db.url,
   ssl: config.db.url.includes('supabase.com') ? { rejectUnauthorized: false } : undefined,
 })
+;(globalThis as any).__lens_pgPool = pgPool
 
 // Instrument raw pg queries
 const originalQuery = pgPool.query.bind(pgPool)
@@ -38,6 +40,7 @@ pgPool.query = (async (...args: any[]) => {
  * stops that reappearing the next time a venue is added.
  */
 export async function upsertPricePoints(points: {
+  id?: string
   assetA: string; assetB: string; pairKey: string; source: string
   poolId?: string; price: number; baseVolume: number; counterVolume: number
   ledger: number; timestamp: Date; eventId?: string
@@ -45,6 +48,7 @@ export async function upsertPricePoints(points: {
   if (points.length === 0) return 0
   const result = await prisma.pricePoint.createMany({
     data: points.map(p => ({
+      id: p.id ?? (p.eventId ? `${network}:${p.source}:${p.eventId}` : undefined),
       network,
       assetA: p.assetA,
       assetB: p.assetB,
diff --git a/src/routes/candles.ts b/src/routes/candles.ts
index c83c56c5..45600053 100644
--- a/src/routes/candles.ts
+++ b/src/routes/candles.ts
@@ -1,5 +1,7 @@
 import type { FastifyInstance } from 'fastify'
 import { pgPool } from '../db'
+import { activeNetwork, type NetworkName } from '../config'
+import { findOverlappingGaps } from '../backfill/gaps'
 
 const INTERVAL_SECONDS: Record<string, number> = {
   '1m': 60,
@@ -56,6 +58,16 @@ export async function registerCandleRoutes(app: FastifyInstance) {
       [intervalSecs, pairKey, from, to]
     )
 
+    const network: NetworkName = (req as any).network ?? activeNetwork
+    const overlappingGaps = await findOverlappingGaps({
+      network,
+      pair: pairKey,
+      start: from,
+      end: to,
+    })
+
+    const gapDeclaration = overlappingGaps.length > 0 ? { incomplete: true, gaps: overlappingGaps } : {}
+
     return {
       pairKey,
       interval,
@@ -69,6 +81,7 @@ export async function registerCandleRoutes(app: FastifyInstance) {
         close: Number(r.close),
         volume: Number(r.volume),
       })),
+      ...gapDeclaration,
     }
   })
 }
diff --git a/src/routes/price.ts b/src/routes/price.ts
index 241a0609..4611f6cf 100644
--- a/src/routes/price.ts
+++ b/src/routes/price.ts
@@ -3,6 +3,7 @@ import { computeTWAP, computeVWAP } from '../pricing/twap'
 import { z } from 'zod'
 import '../middleware/network'
 import { activeNetwork, type NetworkName } from '../config'
+import { findOverlappingGaps } from '../backfill/gaps'
 
 const TwapQuerySchema = z.object({
   window: z.coerce.number().int().min(1).max(1440).default(60),
@@ -53,6 +54,15 @@ export async function registerPriceRoutes(app: FastifyInstance) {
           network,
         })
 
+        const overlappingGaps = await findOverlappingGaps({
+          network,
+          pair: pairKey,
+          start: result.startTime,
+          end: result.endTime,
+        })
+
+        const gapDeclaration = overlappingGaps.length > 0 ? { incomplete: true, gaps: overlappingGaps } : {}
+
         return {
           assetA,
           assetB,
@@ -66,6 +76,7 @@ export async function registerPriceRoutes(app: FastifyInstance) {
           sampleCount: result.sampleCount,
           outlierRejected: result.outlierRejected,
           filterMethod: result.filterMethod,
+          ...gapDeclaration,
         }
       } catch (err) {
         // A driver error message can carry the connection string, and so the
@@ -95,13 +106,25 @@ export async function registerPriceRoutes(app: FastifyInstance) {
       const { window: windowMinutes, source, method } = parsed.data
 
       const pairKey = [assetA, assetB].sort().join('/')
+      const network = (req as any).network ?? activeNetwork
 
       try {
         const result = await computeVWAP(pairKey, windowMinutes, {
           source,
           outlierMethod: method,
+          network,
         })
 
+        const overlappingGaps = await findOverlappingGaps({
+          network,
+          pair: pairKey,
+          start: result.startTime,
+          end: result.endTime,
+          source,
+        })
+
+        const gapDeclaration = overlappingGaps.length > 0 ? { incomplete: true, gaps: overlappingGaps } : {}
+
         return {
           assetA,
           assetB,
@@ -115,6 +138,7 @@ export async function registerPriceRoutes(app: FastifyInstance) {
           volumeTotal: result.volumeTotal,
           outlierRejected: result.outlierRejected,
           filterMethod: result.filterMethod,
+          ...gapDeclaration,
         }
       } catch (err) {
         // Same as TWAP above: the driver message can contain credentials.