Download files

Dataset semantics

Selection and identity

GET /downloads has no query filters. Select data.files locally by vehicle and dataset; archive vehicle is FFMM or FI, with FI covering FIRES and FINRE rows. Use the returned id without constructing ids or paths.

Checkpoint file-level vehicle, dataset, publicationId and sha256. Preserve id, asOf, bytes, rows and columns with the artifact. meta is an aggregate envelope and can combine publications or date cadences; it is not a substitute for file-level identity.

A required file absent from data.files is unavailable; data.pending identifies known publications awaiting artifacts. Neither absence nor pending means an empty dataset. Retain the previous checkpoint instead of committing empty data.

Update policy

publicationId identifies an edition, not a change log. A new edition does not enumerate historical corrections. Use the archive history when full historical consistency is required.

Transfer and commit contract

GET /downloads/{id} returns downloadUrl and expiresAt for the selected active artifact. The URL is a read-only bearer credential, valid for at most 15 minutes. Fetch it without Authorization or the API key, and exclude it from logs and persisted checkpoints.

Match the link response identity to the selected file. Verify bytes and SHA-256 against the compressed payload before decompression or ingestion. On a mismatch, reject the payload and retain the previous checkpoint.

Download state and ingestion state are separate. Persist a download checkpoint only after the verified artifact and metadata have been successfully written. Commit an ingestion checkpoint only after the datastore update succeeds; a downloaded file does not establish successful ingestion.

A batch of N changed files requires one catalog request and N link requests: 1 + N authenticated API requests, plus N direct file GETs. Direct file GETs do not consume API request quota.

Recovery rules

Wire format and keys

UTF-8 CSV inside gzip, with an ordered header, comma delimiters and decimal points. rows excludes the header. Empty fields are null, never zero. Preserve run and serie as strings. NAV ingestion key is run, serie, date; keep currency and netAssetsCurrency independently.

Daily NAV dates use YYYY-MM-DD. Portfolio dates use YYYY-MM. Per-file asOf is a publication cutoff; row dates are observations and can be older. Do not fill missing observations or infer a shared cutoff across vehicles.

Retain dataset and publicationId as row provenance when combining artifacts. Snapshot taxonomy and census facts are not historical replacements; do not backfill earlier observations from them.

NAV column semantics

Column definitions apply to snapshot and history. Use the per-file columns field as the authoritative ordered header. Portfolio has a separate row schema.

Portfolio contract

dataset=portfolio is versioned from its Carteras source and the admitted FFMM set, independently of daily NAV heads. File asOf is the latest included month; every row retains its own asOf. Row fields are run, asOf, source, name, ticker, issuerRun, country, instrumentType and share. share is a proportion: 0.15 means 15%.

Coverage includes admitted FFMM with source holdings, including funds without active NAV. Unknown identities and funds classified as FI at preparation are excluded. Positions are consolidated direct holdings, without ETF or fund look-through. This does not establish final economic exposure.

The individual fund endpoint exposes portfolio.complete and positionCount. positionCount counts source rows before consolidation; complete=false does not establish coverage of omitted positions.

Reference adapter: FFMM archives

Python 3.10+ standard library; FONDOS_API_KEY is the only required input. The adapter archives an initial history file, then snapshots from changed publications. It records compressed bytes and file metadata under fondos-downloads and maintains a download checkpoint in state.json.

The adapter does not schedule runs, retry automatically or update an analytical datastore. Failures emit JSON diagnostics on stderr; HTTP errors include status and numeric Retry-After. URLs and credentials are excluded. For a fresh history baseline, remove fondos-downloads/state.json; existing verified artifacts are retained.

# FFMM archive adapter; entry point: python3 sync_fondos.py
# Python 3.10+ standard library. Required input: FONDOS_API_KEY.
# First run: FFMM history. Later runs: latest FFMM snapshot when changed.
# Snapshots are not a change log. Refresh history for missed dates,
# historical corrections, or newly introduced series: remove state.json.
import hashlib
import json
import os
from pathlib import Path
from urllib.error import HTTPError
from urllib.parse import quote
from urllib.request import Request, urlopen

BASE = "https://ledgifi.com/api/fondos/v1"
ROOT = Path("fondos-downloads")
ROOT.mkdir(exist_ok=True)
STATE = ROOT / "state.json"

def get(path):
    headers = {"Authorization": "Bearer " + os.environ["FONDOS_API_KEY"]}
    with urlopen(Request(BASE + path, headers=headers), timeout=30) as response:
        return json.load(response)["data"]

def save_json(path, value):
    temporary = path.with_suffix(path.suffix + ".tmp")
    temporary.write_text(json.dumps(value, indent=2), encoding="utf-8")
    temporary.replace(path)

def main():
    state = json.loads(STATE.read_text()) if STATE.exists() else {}
    files = get("/downloads")["files"]
    files = {f["dataset"]: f for f in files if f["vehicle"] == "FFMM"}
    dataset = "snapshot" if state else "history"
    if dataset not in files or "snapshot" not in files:
        raise RuntimeError("FFMM download not ready; try again later")
    snapshot = files["snapshot"]
    previous = state.get("snapshot", {})
    if (previous.get("publicationId"), previous.get("sha256")) == (
        snapshot["publicationId"], snapshot["sha256"]
    ):
        print("No new FFMM publication")
        return
    file = files[dataset]
    if file["publicationId"] != snapshot["publicationId"]:
        raise RuntimeError("History and snapshot publication mismatch")
    fields = ("id", "vehicle", "dataset", "publicationId", "asOf", "bytes", "sha256", "columns")
    metadata = {key: file[key] for key in fields}
    link = get("/downloads/" + quote(file["id"], safe=""))
    if any(link[key] != file[key] for key in fields if key != "columns"):
        raise RuntimeError("Download link metadata changed; run again")
    # Keep each publication, even when two versions have identical bytes.
    name = dataset + "-" + quote(file["publicationId"], safe="") + "-" + file["sha256"]
    target = ROOT / (name + ".csv.gz")
    temporary = target.with_suffix(".part")
    digest, size = hashlib.sha256(), 0
    try:
        # This request intentionally has no Authorization header.
        with urlopen(link["downloadUrl"], timeout=120) as response, temporary.open("wb") as output:
            while chunk := response.read(1024 * 1024):
                output.write(chunk)
                digest.update(chunk)
                size += len(chunk)
        if size != file["bytes"] or digest.hexdigest() != file["sha256"]:
            raise RuntimeError("Download integrity check failed")
        temporary.replace(target)
    finally:
        temporary.unlink(missing_ok=True)
    # Raw CSV preserves RUN/serie text, units, and empty values (null).
    # Keep files plus sidecars; downstream upsert key: (run, serie, date).
    save_json(target.with_suffix(".metadata.json"), metadata)
    state[dataset] = metadata
    state["snapshot"] = {key: snapshot[key] for key in fields}
    save_json(STATE, state)
    print("Saved verified FFMM " + dataset)

if __name__ == "__main__":
    try:
        main()
    except HTTPError as error:
        retry_after = error.headers.get("Retry-After", "")
        raise SystemExit(json.dumps({
            "error": "HTTP_ERROR", "status": error.code,
            "retryAfterSeconds": int(retry_after) if retry_after.isdecimal() else None,
        }))
    except Exception:
        # Do not log API keys, signed URLs, or exception request details.
        raise SystemExit(json.dumps({"error": "ADAPTER_ERROR"}))