[verified] add vintage collector and point-in-time API

2026-08-23 11:11:53 +07:00
parent cfa75787bb
commit d1ba6efc68
12 changed files with 589 additions and 13 deletions
--- a/README.md
+++ b/README.md
@@ -59,11 +59,22 @@ Data-health and replay endpoints:

 ```text
 GET /api/v1/data-health
+GET /api/v1/vintages?as_of=<ISO-8601 timestamp>
 GET /api/v1/replay/tourism?vintage_id=<vintage_id>
 ```

 Source: `https://app.bot.or.th/BTWS_STAT/statistics/ReportPage.aspx?reportID=875&language=eng`

+## Collect a vintage manually
+
+The collector is intentionally one-shot and idempotent. Run it after a source update; it preserves raw bytes, normalized snapshots, and a manifest under `backend/data/`:
+
+```bash
+PYTHONPATH=backend .venv/bin/python backend/scripts/collect_tourism_vintage.py --root backend/data
+```
+
+Repeated collection of the same source hash keeps one `vintage_id` and increments `seen_count` without changing `first_seen_at`. A new hash for the same publication timestamp is recorded as a separate `revised` vintage.
+
 ## Tests and build

 ```bash
@@ -71,18 +82,19 @@ PYTHONPATH=backend .venv/bin/python -m unittest discover -s backend/tests -v
 cd frontend && npm run build
 ```

-## Current M1 boundary
+## Current M2 boundary

 - English UI and analysis vocabulary
 - Research mode and paper mode only
 - Tourism Pulse fixture adapter and BOT Tourism Indicators adapter
 - Data lineage: source, publication time, retrieval time, vintage
 - Raw response hash and normalized snapshot persistence
- Read-only data-health and vintage replay endpoints
+- Immutable vintage manifest with first-seen/revision metadata
+- Read-only data-health, vintage timeline and vintage replay endpoints
 - Deterministic surprise × exposure × confidence score
 - Paper ledger endpoint
 - No LLM call yet; the deterministic result is the source of truth
 - No webhook receiver yet
 - No MT5 bridge yet

-The next implementation step is to validate multiple BOT vintages and add the next real Tourism metric before introducing LLM analysis.
+The next implementation step is the event-study/backtest layer using only vintages whose `published_at` is known at each test date.
--- a/backend/app/init.py
+++ b/backend/app/init.py
@@ -16,8 +16,9 @@ from flask import Flask, jsonify, request
 from .bot_tourism import BotTourismSource, TourismSourceError
 from .paper import PaperLedger
 from .tourism import compute_tourism_signal
+from .vintages import VintageStore, VintageStoreError

-APP_VERSION = "0.2.0"
+APP_VERSION = "0.3.0"


 def _load_default_snapshot() -> dict[str, Any]:
@@ -46,20 +47,21 @@ def create_app(config: dict[str, Any] | None = None) -> Flask:
        PAPER_SESSION_SECONDS=int(os.getenv("PAPER_SESSION_SECONDS", "3600")),
        TOURISM_SOURCE=os.getenv("TOURISM_SOURCE", "fixture"),
        TOURISM_ADAPTER=None,
+        TOURISM_DATA_ROOT=Path(os.getenv("TOURISM_DATA_ROOT", str(data_root))),
        TOURISM_RAW_DIR=Path(os.getenv("TOURISM_RAW_DIR", str(data_root / "raw" / "tourism"))),
        SNAPSHOT_DIR=Path(os.getenv("TOURISM_SNAPSHOT_DIR", str(data_root / "snapshots"))),
+        VINTAGE_STORE=None,
        SNAPSHOT=_load_default_snapshot(),
    )
    if config:
        app.config.update(config)

+    vintage_store = app.config.get("VINTAGE_STORE") or VintageStore(app.config["TOURISM_DATA_ROOT"])
+    app.extensions["vintage_store"] = vintage_store
    source_snapshot = app.config["SNAPSHOT"]
    source_mode = str(app.config["TOURISM_SOURCE"]).lower()
    if source_mode == "bot":
-        adapter = app.config.get("TOURISM_ADAPTER") or BotTourismSource(
-            raw_dir=Path(app.config["TOURISM_RAW_DIR"]),
-            snapshot_dir=Path(app.config["SNAPSHOT_DIR"]),
-        )
+        adapter = app.config.get("TOURISM_ADAPTER") or BotTourismSource(vintage_store=vintage_store)
        try:
            source_snapshot = adapter.fetch(exposures=source_snapshot.get("exposures", []))
        except TourismSourceError as exc:
@@ -164,6 +166,15 @@ def create_app(config: dict[str, Any] | None = None) -> Flask:
            }
        )

+    @app.get("/api/v1/vintages")
+    def vintages():
+        as_of = request.args.get("as_of")
+        try:
+            entries = app.extensions["vintage_store"].list_vintages(as_of)
+        except VintageStoreError as exc:
+            return jsonify({"error": str(exc)}), 400
+        return jsonify({"as_of": as_of, "count": len(entries), "vintages": entries})
+
    @app.get("/api/v1/replay/tourism")
    def replay_tourism():
        vintage_id = request.args.get("vintage_id", "")
--- a/backend/app/bot_tourism.py
+++ b/backend/app/bot_tourism.py
@@ -19,6 +19,8 @@ from urllib.request import HTTPCookieProcessor, Request, build_opener

 from http.cookiejar import CookieJar

+from .vintages import VintageStore
+
 BOT_TOURISM_URL = "https://app.bot.or.th/BTWS_STAT/statistics/ReportPage.aspx?reportID=875&language=eng"
 BOT_SOURCE_ID = "bot.ec_ei_028_s2"
 PARSER_VERSION = "bot-tourism-v1"
@@ -273,6 +275,7 @@ class BotTourismSource:
        clock: Callable[[], datetime] | None = None,
        raw_dir: Path | None = None,
        snapshot_dir: Path | None = None,
+        vintage_store: VintageStore | None = None,
    ) -> None:
        self.url = url
        self.timeout = timeout
@@ -280,6 +283,7 @@ class BotTourismSource:
        self.clock = clock or (lambda: datetime.now(timezone.utc))
        self.raw_dir = raw_dir
        self.snapshot_dir = snapshot_dir
+        self.vintage_store = vintage_store

    def _open(self, request: Request) -> bytes:
        try:
@@ -336,6 +340,8 @@ class BotTourismSource:
            source_url=self.url,
            exposures=exposures,
        )
+        if self.vintage_store is not None:
+            return self.vintage_store.persist(report_bytes, snapshot)
        source_meta = snapshot["source"]
        if self.raw_dir is not None:
            _atomic_write(self.raw_dir / source_meta["raw_snapshot_file"], report_bytes)
--- a/backend/app/collector.py
+++ b/backend/app/collector.py
@@ -0,0 +1,45 @@
+"""One-shot Tourism vintage collector."""
+
+from __future__ import annotations
+
+import json
+from datetime import datetime
+from pathlib import Path
+from typing import Any, Callable, Iterable
+
+from .bot_tourism import BotTourismSource
+from .vintages import VintageStore
+
+
+def _default_exposures() -> list[dict[str, Any]]:
+    fixture_path = Path(__file__).resolve().parents[1] / "fixtures" / "tourism_snapshot.json"
+    snapshot = json.loads(fixture_path.read_text(encoding="utf-8"))
+    return [dict(item) for item in snapshot.get("exposures", [])]
+
+
+def collect_bot_vintage(
+    root: Path | str,
+    *,
+    opener: Any | None = None,
+    clock: Callable[[], datetime] | None = None,
+    timeout: float = 30.0,
+    exposures: Iterable[dict[str, Any]] | None = None,
+) -> dict[str, Any]:
+    """Fetch one BOT vintage and register it in the immutable manifest."""
+
+    store = VintageStore(root)
+    source = BotTourismSource(
+        opener=opener,
+        clock=clock,
+        timeout=timeout,
+        vintage_store=store,
+    )
+    snapshot = source.fetch(exposures=_default_exposures() if exposures is None else exposures)
+    vintage_id = snapshot["source"]["vintage_id"]
+    entry = store.load_manifest()["vintages"][vintage_id]
+    return {
+        "snapshot": snapshot,
+        "manifest_path": str(store.manifest_path),
+        "seen_count": entry["seen_count"],
+        "revision_status": entry["revision_status"],
+    }
--- a/backend/app/vintages.py
+++ b/backend/app/vintages.py
@@ -0,0 +1,167 @@
+"""Immutable tourism vintage manifest and point-in-time helpers."""
+
+from __future__ import annotations
+
+import copy
+import hashlib
+import json
+import re
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any, Iterable
+
+MANIFEST_SCHEMA_VERSION = 1
+_VINTAGE_ID_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9._-]{0,127}")
+_FILENAME_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9._-]{0,191}\.(?:html|json)")
+
+
+class VintageStoreError(ValueError):
+    """Raised when a vintage cannot be safely stored or loaded."""
+
+
+def _parse_timestamp(value: str) -> datetime:
+    try:
+        parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
+    except (TypeError, ValueError) as exc:
+        raise VintageStoreError("timestamp must be ISO-8601") from exc
+    if parsed.tzinfo is None:
+        parsed = parsed.replace(tzinfo=timezone.utc)
+    return parsed
+
+
+def _safe_vintage_id(vintage_id: str) -> str:
+    if not _VINTAGE_ID_RE.fullmatch(vintage_id):
+        raise VintageStoreError("invalid vintage_id")
+    return vintage_id
+
+
+def _safe_filename(filename: str, suffix: str) -> str:
+    if not _FILENAME_RE.fullmatch(filename) or not filename.endswith(suffix):
+        raise VintageStoreError("invalid vintage filename")
+    if Path(filename).name != filename:
+        raise VintageStoreError("vintage filename must not contain a path")
+    return filename
+
+
+def filter_vintages(entries: Iterable[dict[str, Any]], as_of: str | None = None) -> list[dict[str, Any]]:
+    """Return vintages whose publication timestamp was known by ``as_of``."""
+
+    cutoff = _parse_timestamp(as_of) if as_of else None
+    selected = []
+    for entry in entries:
+        published_at = _parse_timestamp(str(entry["published_at"]))
+        if cutoff is None or published_at <= cutoff:
+            selected.append(copy.deepcopy(entry))
+    return sorted(selected, key=lambda item: (_parse_timestamp(str(item["published_at"])), item["vintage_id"]))
+
+
+def _atomic_write_bytes(path: Path, content: bytes) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    temporary = path.with_name(f".{path.name}.tmp")
+    temporary.write_bytes(content)
+    temporary.replace(path)
+
+
+def _atomic_write_json(path: Path, payload: dict[str, Any]) -> None:
+    _atomic_write_bytes(path, (json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True) + "\n").encode("utf-8"))
+
+
+class VintageStore:
+    """Filesystem-backed immutable raw/snapshot store with a compact manifest."""
+
+    def __init__(self, root: Path | str) -> None:
+        self.root = Path(root).resolve()
+        self.raw_dir = self.root / "raw" / "tourism"
+        self.snapshot_dir = self.root / "snapshots"
+        self.manifest_path = self.root / "manifest.json"
+
+    def load_manifest(self) -> dict[str, Any]:
+        if not self.manifest_path.is_file():
+            return {"schema_version": MANIFEST_SCHEMA_VERSION, "vintages": {}}
+        try:
+            payload = json.loads(self.manifest_path.read_text(encoding="utf-8"))
+        except (OSError, json.JSONDecodeError) as exc:
+            raise VintageStoreError("vintage manifest is unreadable") from exc
+        if not isinstance(payload, dict) or payload.get("schema_version") != MANIFEST_SCHEMA_VERSION:
+            raise VintageStoreError("unsupported vintage manifest schema")
+        if not isinstance(payload.get("vintages"), dict):
+            raise VintageStoreError("vintage manifest has invalid vintages map")
+        return payload
+
+    def persist(self, raw_bytes: bytes, snapshot: dict[str, Any]) -> dict[str, Any]:
+        if not isinstance(raw_bytes, bytes) or not raw_bytes:
+            raise VintageStoreError("raw vintage payload must be non-empty bytes")
+        stored_snapshot = copy.deepcopy(snapshot)
+        source = stored_snapshot.get("source")
+        if not isinstance(source, dict):
+            raise VintageStoreError("snapshot source metadata is required")
+        vintage_id = _safe_vintage_id(str(source.get("vintage_id", "")))
+        published_at = str(source.get("published_at", ""))
+        retrieved_at = str(source.get("retrieved_at", ""))
+        _parse_timestamp(published_at)
+        _parse_timestamp(retrieved_at)
+        raw_hash = hashlib.sha256(raw_bytes).hexdigest()
+        declared_hash = source.get("raw_payload_hash")
+        if declared_hash and declared_hash != raw_hash:
+            raise VintageStoreError("raw payload hash does not match snapshot metadata")
+        source["raw_payload_hash"] = raw_hash
+        source["raw_snapshot_file"] = _safe_filename(
+            str(source.get("raw_snapshot_file") or f"{vintage_id}.html"), ".html"
+        )
+        source["snapshot_file"] = _safe_filename(
+            str(source.get("snapshot_file") or f"{vintage_id}.json"), ".json"
+        )
+
+        manifest = self.load_manifest()
+        existing = manifest["vintages"].get(vintage_id)
+        same_release = any(
+            item.get("source_id") == source.get("source_id")
+            and item.get("published_at") == published_at
+            and item.get("vintage_id") != vintage_id
+            for item in manifest["vintages"].values()
+        )
+        revision_status = existing.get("revision_status") if existing else ("revised" if same_release else "initial")
+        source["revision_status"] = revision_status
+        stored_snapshot["source"] = source
+
+        _atomic_write_bytes(self.raw_dir / source["raw_snapshot_file"], raw_bytes)
+        _atomic_write_json(self.snapshot_dir / source["snapshot_file"], stored_snapshot)
+
+        first_seen_at = existing.get("first_seen_at", retrieved_at) if existing else retrieved_at
+        last_seen_at = max(_parse_timestamp(first_seen_at), _parse_timestamp(retrieved_at)).isoformat()
+        manifest["vintages"][vintage_id] = {
+            "vintage_id": vintage_id,
+            "source_id": source.get("source_id"),
+            "source_url": source.get("source_url"),
+            "as_of": stored_snapshot.get("as_of"),
+            "published_at": published_at,
+            "release_status": source.get("release_status"),
+            "revision_status": revision_status,
+            "raw_payload_hash": raw_hash,
+            "parser_version": source.get("parser_version"),
+            "raw_snapshot_file": source["raw_snapshot_file"],
+            "snapshot_file": source["snapshot_file"],
+            "first_seen_at": first_seen_at,
+            "last_seen_at": last_seen_at,
+            "seen_count": int(existing.get("seen_count", 0)) + 1 if existing else 1,
+        }
+        _atomic_write_json(self.manifest_path, manifest)
+        return stored_snapshot
+
+    def list_vintages(self, as_of: str | None = None) -> list[dict[str, Any]]:
+        return filter_vintages(self.load_manifest()["vintages"].values(), as_of)
+
+    def load_snapshot(self, vintage_id: str) -> dict[str, Any]:
+        vintage_id = _safe_vintage_id(vintage_id)
+        snapshot_path = (self.snapshot_dir / f"{vintage_id}.json").resolve()
+        if snapshot_path.parent != self.snapshot_dir.resolve():
+            raise VintageStoreError("snapshot path escaped store root")
+        if not snapshot_path.is_file():
+            raise FileNotFoundError(vintage_id)
+        try:
+            snapshot = json.loads(snapshot_path.read_text(encoding="utf-8"))
+        except (OSError, json.JSONDecodeError) as exc:
+            raise VintageStoreError("vintage snapshot is unreadable") from exc
+        if snapshot.get("source", {}).get("vintage_id") != vintage_id:
+            raise VintageStoreError("vintage snapshot identity mismatch")
+        return snapshot
--- a/backend/scripts/collect_tourism_vintage.py
+++ b/backend/scripts/collect_tourism_vintage.py
@@ -0,0 +1,44 @@
+"""Collect one live BOT Tourism vintage into the local manifest."""
+
+from __future__ import annotations
+
+import argparse
+import json
+from pathlib import Path
+
+from app.collector import collect_bot_vintage
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument(
+        "--root",
+        type=Path,
+        default=Path(__file__).resolve().parents[1] / "data",
+        help="vintage store root (default: backend/data)",
+    )
+    parser.add_argument("--timeout", type=float, default=30.0)
+    args = parser.parse_args()
+    result = collect_bot_vintage(args.root, timeout=args.timeout)
+    snapshot = result["snapshot"]
+    print(
+        json.dumps(
+            {
+                "source_id": snapshot["source"]["source_id"],
+                "vintage_id": snapshot["source"]["vintage_id"],
+                "as_of": snapshot["as_of"],
+                "published_at": snapshot["source"]["published_at"],
+                "release_status": snapshot["source"]["release_status"],
+                "seen_count": result["seen_count"],
+                "revision_status": result["revision_status"],
+                "manifest_path": result["manifest_path"],
+            },
+            ensure_ascii=False,
+            sort_keys=True,
+        )
+    )
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
--- a/backend/tests/test_api.py
+++ b/backend/tests/test_api.py
@@ -4,6 +4,7 @@ import unittest
 from pathlib import Path

 from app import create_app
+from app.vintages import VintageStore


 class ApiTests(unittest.TestCase):
@@ -94,6 +95,23 @@ class ApiTests(unittest.TestCase):
        self.assertEqual(body["source_id"], "bot.ec_ei_028_s2")
        self.assertEqual(body["status"], "provisional")

+    def test_vintages_endpoint_filters_by_publication_timestamp(self):
+        with tempfile.TemporaryDirectory() as temp_dir:
+            store = VintageStore(Path(temp_dir))
+            store.persist(b"fixture raw", self.snapshot)
+            app = create_app({"TESTING": True, "SNAPSHOT": self.snapshot, "VINTAGE_STORE": store})
+            client = app.test_client()
+            before = client.get("/api/v1/vintages?as_of=2026-08-20T00:00:00Z")
+            after = client.get("/api/v1/vintages?as_of=2026-08-22T00:00:00Z")
+            self.assertEqual(before.status_code, 200)
+            self.assertEqual(before.get_json()["count"], 0)
+            self.assertEqual(after.status_code, 200)
+            self.assertEqual(after.get_json()["count"], 1)
+
+    def test_vintages_endpoint_rejects_invalid_as_of(self):
+        response = self.client.get("/api/v1/vintages?as_of=not-a-timestamp")
+        self.assertEqual(response.status_code, 400)
+
    def test_paper_ledger_requires_token(self):
        response = self.client.post(
            "/api/v1/paper/ledger",
--- a/backend/tests/test_vintages.py
+++ b/backend/tests/test_vintages.py
@@ -0,0 +1,131 @@
+import copy
+import hashlib
+import json
+import tempfile
+import unittest
+from datetime import datetime, timezone
+from pathlib import Path
+from urllib.parse import parse_qs
+
+from app.bot_tourism import BotTourismSource
+from app.collector import collect_bot_vintage
+from app.vintages import VintageStore, filter_vintages
+
+FIXTURES = Path(__file__).parent / "fixtures"
+REPORT_HTML = (FIXTURES / "bot_tourism_report.html").read_text(encoding="utf-8")
+FORM_HTML = (FIXTURES / "bot_tourism_form.html").read_text(encoding="utf-8")
+
+
+class FakeResponse:
+    def __init__(self, body):
+        self.body = body.encode("utf-8")
+        self.status = 200
+
+    def read(self):
+        return self.body
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *_args):
+        return False
+
+
+class FakeOpener:
+    def __init__(self):
+        self.calls = []
+
+    def open(self, request, timeout):
+        body = FORM_HTML if request.data is None else REPORT_HTML
+        self.calls.append({"url": request.full_url, "body": request.data, "timeout": timeout})
+        return FakeResponse(body)
+
+
+def sample_snapshot(retrieved_at="2026-08-23T01:00:00+00:00"):
+    from app.bot_tourism import parse_bot_tourism_report
+
+    return parse_bot_tourism_report(REPORT_HTML, retrieved_at=retrieved_at, exposures=[])
+
+
+class VintageStoreTests(unittest.TestCase):
+    def test_manifest_preserves_first_seen_and_deduplicates_same_vintage(self):
+        with tempfile.TemporaryDirectory() as temp_dir:
+            store = VintageStore(Path(temp_dir))
+            first = sample_snapshot("2026-08-23T01:00:00+00:00")
+            store.persist(REPORT_HTML.encode("utf-8"), first)
+            second = copy.deepcopy(first)
+            second["source"]["retrieved_at"] = "2026-08-24T01:00:00+00:00"
+            store.persist(REPORT_HTML.encode("utf-8"), second)
+
+            manifest = store.load_manifest()
+            self.assertEqual(len(manifest["vintages"]), 1)
+            entry = next(iter(manifest["vintages"].values()))
+            self.assertEqual(entry["first_seen_at"], "2026-08-23T01:00:00+00:00")
+            self.assertEqual(entry["last_seen_at"], "2026-08-24T01:00:00+00:00")
+            self.assertEqual(entry["seen_count"], 2)
+            self.assertEqual(entry["revision_status"], "initial")
+
+    def test_manifest_marks_same_release_with_new_hash_as_revision(self):
+        with tempfile.TemporaryDirectory() as temp_dir:
+            store = VintageStore(Path(temp_dir))
+            first = sample_snapshot()
+            store.persist(REPORT_HTML.encode("utf-8"), first)
+            revised = copy.deepcopy(first)
+            revised_raw = b"revised raw"
+            revised_hash = hashlib.sha256(revised_raw).hexdigest()
+            revised["source"]["raw_payload_hash"] = revised_hash
+            revised["source"]["vintage_id"] = revised["source"]["vintage_id"][:-12] + revised_hash[:12]
+            revised["source"]["raw_snapshot_file"] = revised["source"]["vintage_id"] + ".html"
+            revised["source"]["snapshot_file"] = revised["source"]["vintage_id"] + ".json"
+            store.persist(revised_raw, revised)
+
+            entries = list(store.load_manifest()["vintages"].values())
+            self.assertEqual(len(entries), 2)
+            self.assertEqual(sorted(entry["revision_status"] for entry in entries), ["initial", "revised"])
+
+    def test_point_in_time_filter_excludes_future_publications(self):
+        entries = [
+            {"vintage_id": "old", "published_at": "2026-07-31T14:30:00+07:00"},
+            {"vintage_id": "future", "published_at": "2026-08-31T14:30:00+07:00"},
+        ]
+        result = filter_vintages(entries, "2026-08-01T00:00:00+00:00")
+        self.assertEqual([entry["vintage_id"] for entry in result], ["old"])
+
+    def test_snapshot_load_rejects_path_traversal(self):
+        with tempfile.TemporaryDirectory() as temp_dir:
+            store = VintageStore(Path(temp_dir))
+            with self.assertRaises(ValueError):
+                store.load_snapshot("../secret")
+
+    def test_collector_returns_manifested_snapshot_summary(self):
+        with tempfile.TemporaryDirectory() as temp_dir:
+            result = collect_bot_vintage(
+                Path(temp_dir),
+                opener=FakeOpener(),
+                clock=lambda: datetime(2026, 8, 23, 2, 0, tzinfo=timezone.utc),
+            )
+            snapshot = result["snapshot"]
+            self.assertEqual(result["seen_count"], 1)
+            self.assertTrue(Path(result["manifest_path"]).is_file())
+            self.assertEqual(snapshot["source"]["source_id"], "bot.ec_ei_028_s2")
+
+        with tempfile.TemporaryDirectory() as temp_dir:
+            root = Path(temp_dir)
+            store = VintageStore(root)
+            source = BotTourismSource(
+                opener=FakeOpener(),
+                clock=lambda: datetime(2026, 8, 23, 2, 0, tzinfo=timezone.utc),
+                vintage_store=store,
+            )
+            snapshot = source.fetch()
+            manifest = store.load_manifest()
+            self.assertEqual(len(manifest["vintages"]), 1)
+            self.assertEqual(manifest["vintages"][snapshot["source"]["vintage_id"]]["seen_count"], 1)
+            loaded = store.load_snapshot(snapshot["source"]["vintage_id"])
+            self.assertEqual(loaded["source"]["vintage_id"], snapshot["source"]["vintage_id"])
+            self.assertTrue((root / "raw" / "tourism").exists())
+            self.assertTrue((root / "snapshots").exists())
+
+
+if __name__ == "__main__":
+    unittest.main()
--- a/docs/HANDOFF.md
+++ b/docs/HANDOFF.md
@@ -5,10 +5,10 @@
 - Path: `/Users/kunthawat/Gitea/set50-alternative-data-platform`
 - Branch: `main`
 - Verified code commit: `a113a51` — `[verified] add BOT tourism source adapter`
- Current milestone: M1 complete and independently reviewed
+- Current milestone: M2 vintage foundation complete; event study deferred
 - Mode: research + paper only
 - Frontend: Vue 3 + Vite
- Backend: Flask `0.2.0`
+- Backend: Flask `0.3.0`
 - Current runtime source: BOT Tourism Indicators (`TOURISM_SOURCE=bot`)

 ## Completed
@@ -19,6 +19,9 @@
 - Atomic raw HTML and normalized snapshot persistence under ignored `backend/data/`.
 - Read-only `/api/v1/data-health` endpoint.
 - Safe `/api/v1/replay/tourism?vintage_id=...` endpoint with identity and path validation.
+- Immutable `VintageStore` manifest with first-seen, last-seen, seen-count and revision metadata.
+- One-shot collector: `backend/scripts/collect_tourism_vintage.py`.
+- Point-in-time vintage query: `GET /api/v1/vintages?as_of=<ISO-8601>`.
 - Ranked target weights and LONG/SHORT/NEUTRAL classification.
 - English dashboard with live/provisional source label, sign-aware surprise copy and lineage fields.
 - HttpOnly paper session and internal paper ledger.
@@ -50,13 +53,19 @@ npm run build
 Vite build completed successfully.

 GET /api/v1/health
-HTTP 200; {"mode":"research","status":"ok","version":"0.2.0"}
+HTTP 200; {"mode":"research","status":"ok","version":"0.3.0"}

 GET /api/v1/data-health
 HTTP 200; source_mode=bot, status=provisional, replayable=true

 GET /api/v1/replay/tourism?vintage_id=bot.ec_ei_028_s2-2026-07-31-665981f88b4a
 HTTP 200; replay theme surprise matched live summary exactly.
+
+GET /api/v1/vintages?as_of=2026-07-01T00:00:00Z
+HTTP 200; count=0
+
+GET /api/v1/vintages?as_of=2026-08-01T00:00:00Z
+HTTP 200; count=1; manifest seen_count=4
 ```

 Paper writes use a server-side token exchange and HttpOnly `paper_session` cookie; the token is not embedded in the frontend bundle.
@@ -67,8 +76,9 @@ Paper writes use a server-side token exchange and HttpOnly `paper_session` cooki
 - Only foreign-arrival YoY is live in this slice; occupancy and airport passenger metrics are not yet connected.
 - Snapshot storage is local filesystem and single-process; shared persistence is required before multi-worker deployment.
 - No investment edge, transaction-cost result, or backtest conclusion has been established.
+- One independent source release is not enough for a valid event study; current historical rows are not treated as point-in-time vintages.
 - Browser screenshot verification remains blocked by the Chrome remote-debugging permission prompt; served HTML/source, live API, fresh Vite build and replay integrity were verified instead.

 ## Exact next action

-Validate multiple BOT vintages and add the next real Tourism metric without changing the snapshot contract. Run event-study/backtest checks before adding LLM analysis.
+Collect independent BOT releases over time, then implement event-study/backtest checks using only vintages whose `published_at` is known at each test date. Add another metric only when its historical release coverage is real.
--- a/docs/engineering-log.md
+++ b/docs/engineering-log.md
@@ -6,6 +6,7 @@
 |---|---|---|---|
 | M0 repo foundation | complete | Flask API, Vue/Vite shell | keep research/paper guardrails |
 | M1 BOT Tourism adapter | complete | 18 tests, live BOT fetch, raw/snapshot persistence | validate multiple vintages |
+| M2 vintage collector | complete | 25 tests, manifest idempotency, live collector and point-in-time API | collect independent releases |
 | Tourism deterministic signal | complete | live foreign-arrivals YoY surprise | add occupancy/airport metric |
 | Internal paper ledger | complete | POST/readback through live API | persist in PostgreSQL later |
 | Dashboard | complete | Vite build + served source check with live-sign copy | visual browser capture after permission is available |
@@ -23,7 +24,7 @@

 ## Verification

- Backend: 18 unittest tests pass.
+- Backend: 25 unittest tests pass.
 - Independent M1 review: **PASSED**; no concrete security or logic blockers.
 - Reviewer suggestions: set `PAPER_COOKIE_SECURE=1` outside local HTTP; replace in-memory sessions before multi-worker deployment.
 - M1 reviewer backlog: add schema-drift, duplicate/reordered-row, and malformed-vintage regression fixtures.
@@ -34,4 +35,6 @@
 - BOT source live fetch parsed 138 monthly periods and persisted raw HTML plus normalized snapshot.
 - Data-health reports `source_mode=bot`, `status=provisional`, and `replayable=true`.
 - Live vintage replay returned the same theme surprise as the current dashboard summary.
+- Vintage collector preserved one live `vintage_id` with `seen_count=4` and point-in-time API excluded it before `published_at`.
+- Independent M2 review: **PASSED**; no concrete security or logic blockers.
 - Browser visual capture was blocked by Chrome remote-debugging permission; no permission dialog was clicked.
--- a/docs/engineering-log/2026-08-23-vintage-collector.md
+++ b/docs/engineering-log/2026-08-23-vintage-collector.md
@@ -0,0 +1,82 @@
+# 2026-08-23 — M2 vintage collector and point-in-time query
+
+## Plan status
+
+- Vintage collector: complete.
+- Immutable manifest: complete.
+- Point-in-time filtering API: complete.
+- Event study/backtest: deferred until enough independently collected releases exist.
+
+## Changed files
+
+- `backend/app/vintages.py` — manifest schema, safe snapshot store, first/last seen metadata, revision classification, and publication-time filtering.
+- `backend/app/collector.py` — one-shot collector function using the BOT adapter and default exposure matrix.
+- `backend/scripts/collect_tourism_vintage.py` — CLI entrypoint.
+- `backend/app/bot_tourism.py` — optional `VintageStore` persistence path.
+- `backend/app/__init__.py` — shared store wiring and `/api/v1/vintages` endpoint.
+- `backend/tests/test_vintages.py` — manifest, idempotency, revision, traversal, collector and point-in-time tests.
+- `backend/tests/test_api.py` — vintage API filter and invalid timestamp tests.
+- `README.md` — collector and M2 boundary documentation.
+
+## Manifest contract
+
+Each entry stores:
+
+```text
+vintage_id
+source_id
+source_url
+as_of
+published_at
+release_status
+revision_status
+raw_payload_hash
+parser_version
+raw_snapshot_file
+snapshot_file
+first_seen_at
+last_seen_at
+seen_count
+```
+
+Repeated collection of the same raw hash keeps one vintage and increments `seen_count`. A different hash for the same source publication timestamp becomes a separate `revised` vintage.
+
+## Live evidence
+
+The real collector ran twice manually and the app fetched once on restart. The manifest contains one BOT vintage with `seen_count=4`, `revision_status=initial`, `published_at=2026-07-31T14:30:00+07:00`, and the same raw hash `665981f88b4a30c5bd30026cf1e96279c244ad83725558f4136d952b29756a31`.
+
+Point-in-time API checks:
+
+```text
+GET /api/v1/vintages
+count: 1
+
+GET /api/v1/vintages?as_of=2026-07-01T00:00:00Z
+count: 0
+
+GET /api/v1/vintages?as_of=2026-08-01T00:00:00Z
+count: 1
+```
+
+## Verification
+
+- `PYTHONPATH=backend .venv/bin/python -W error -m unittest discover -s backend/tests -v` — **25 tests passed**.
+- `npm run build` — passed.
+- `npm audit --omit=dev --audit-level=high` — 0 vulnerabilities.
+- `GET /api/v1/health` — HTTP 200, version `0.3.0`.
+- `GET /api/v1/data-health` — HTTP 200, `source_mode=bot`, `replayable=true`.
+- Collector and API used the same manifest/snapshot root.
+
+## Independent review
+
+```text
+passed: true
+security_concerns: []
+logic_errors: []
+```
+
+Non-blocking backlog: add concurrent-writer/restart recovery coverage and more timezone/revision edge cases.
+
+## Risks and next action
+
+One collected release is not enough for an event study. The current historical series is not treated as historical point-in-time knowledge. Continue collecting new releases, then implement event windows and cost-aware backtesting only over vintages whose publication timestamps are known.
--- a/docs/test-evidence/2026-08-23-vintage-collector.md
+++ b/docs/test-evidence/2026-08-23-vintage-collector.md
@@ -0,0 +1,47 @@
+# Test evidence — 2026-08-23 vintage collector
+
+## Automated
+
+```text
+PYTHONPATH=backend .venv/bin/python -W error -m unittest discover -s backend/tests -v
+Ran 25 tests ... OK
+
+npm run build
+Vite build completed successfully.
+
+npm audit --omit=dev --audit-level=high
+found 0 vulnerabilities
+```
+
+## Live collector
+
+The one-shot collector ran twice against the BOT source and the application fetched once on restart. Results:
+
+```text
+manifest vintages: 1
+seen_count: 4
+revision_status: initial
+source_id: bot.ec_ei_028_s2
+published_at: 2026-07-31T14:30:00+07:00
+```
+
+## Point-in-time API
+
+```text
+as_of=2026-07-01T00:00:00Z → count 0
+as_of=2026-08-01T00:00:00Z → count 1
+```
+
+## Manual visual/network limitation
+
+The frontend build and served Vite source were previously verified. Browser screenshot capture remains unavailable because the Chrome remote-debugging permission prompt has not been approved. Backend live endpoints were verified over HTTP.
+
+## Independent review
+
+```text
+passed: true
+security_concerns: []
+logic_errors: []
+```
+
+Non-blocking backlog: add concurrent-writer/restart recovery coverage and more timezone/revision edge cases.