Shinkansen · Treasury (API → BigQuery)

De punta a punta: una extracción incremental diaria desde la Treasury API de Shinkansen (/v1/payout_orders) que termina en la tabla cruda shinkansen.shinkansen_treasury en BigQuery. A diferencia de metadata/results, no es un webhook: es un Cloud Run que llama a la API y publica a Pub/Sub.

¿Llegaste hasta acá desde…?

Usa esta tabla para entrar en sentido inverso (lo que ves → de dónde sale):

Si ves…	Eso sale de…	La lógica está en…
La tabla `shinkansen.shinkansen_treasury` en BigQuery / Metabase	`solvento-data-prod.shinkansen.shinkansen_treasury` (RAW)	Ingesta (sin lógica de negocio); ver "Paso a paso"
La extracción incremental (qué fecha trae)	Endpoint `/extract-incremental/treasury` del Cloud Run	`api_shinkansen_handler` → `cloud_run/main.py` (consulta `MAX(inserted_at)` en BQ y arma `gte_inserted_at`)

Resumen

Campo	Valor
Origen	Treasury API de Shinkansen (`https://api.treasury.shinkansen.finance/v1/payout_orders`)
Patrón	API pull incremental (Cloud Scheduler → Cloud Run → Pub/Sub → Dataflow → BigQuery)
Tabla RAW	`solvento-data-prod.shinkansen.shinkansen_treasury`
Frecuencia	Diaria · `0 4 * * *` (America/Mexico_City), job `shinkansen-treasury-api-daily`
Criticidad	🟡 Media
Dueño	Martin (data eng)
Repos	`api_shinkansen_handler` · `templates-dataflow`

Diagrama de linaje

graph LR
    SCH[["⏰ Cloud Scheduler<br/>shinkansen-treasury-api-daily"]]
    CR["Cloud Run<br/>treasury-api-shinkansen-requests-handler-prod<br/><i>/extract-incremental/treasury</i>"]
    API[["🌐 Treasury API Shinkansen<br/>/v1/payout_orders"]]
    BQQ[("BQ lookup<br/>MAX(inserted_at)")]
    TOP(["Pub/Sub<br/>treasury-api-shinkansen-success"])
    SUB(["sub<br/>treasury-api-shinkansen-success-sub"])
    DF["Dataflow<br/>treasury-shinkansen-api-prod-20260120-110700"]
    BQ[("solvento-data-prod<br/>shinkansen.shinkansen_treasury")]
    MB["Metabase / análisis"]

    SCH -->|"dispara diario"| CR
    CR -->|"1· lee última fecha"| BQQ
    CR -->|"2· GET incremental (gte_inserted_at)"| API
    API -->|"payout_orders"| CR
    CR -->|"3· publica por página"| TOP --> SUB --> DF --> BQ --> MB

    classDef src fill:#fde68a,stroke:#b45309,color:#000;
    classDef store fill:#bfdbfe,stroke:#1e40af,color:#000;
    class SCH,API src;
    class BQ store;

Paso a paso

Cloud Scheduler shinkansen-treasury-api-daily dispara diariamente el endpoint /extract-incremental/treasury del Cloud Run.
Cloud Run treasury-api-shinkansen-requests-handler-prod (api_shinkansen_handler, cloud_run/main.py):
- Lee la última fecha cargada: SELECT MAX(inserted_at) FROM solvento-data-prod.shinkansen.shinkansen_treasury (BIGQUERY_DATASET=shinkansen, BIGQUERY_TABLE_TREASURY=shinkansen_treasury, BIGQUERY_DATE_COLUMN=inserted_at).
- Resta un margen (INCREMENTAL_LOOKBACK_DAYS, default 1) y arma la URL incremental con gte_inserted_at contra /v1/payout_orders.
- Llama a la Treasury API (make_api_request, header Shinkansen-Treasury-Api-Key), paginando la respuesta (meta.total_pages).
- Publica cada página (el array payout_orders) en el topic treasury-api-shinkansen-success (OUTPUT_TOPIC_NAME). Los fallos se publican en treasury-api-shinkansen-errors y se reintentan vía Cloud Tasks (queue treasury-api-shinkansen-queue, endpoint /process-task).
Pub/Sub desacopla: topic treasury-api-shinkansen-success → suscripción treasury-api-shinkansen-success-sub.
Dataflow treasury-shinkansen-api-prod-20260120-110700 (templates-dataflow, carpeta shinkansen-treasury-api; INPUT_SUBSCRIPTION=…/treasury-api-shinkansen-success-sub, template dataflow_template_shinkansen.py, BIGQUERY_DATASET=shinkansen) consume la suscripción.
BigQuery: el Dataflow escribe el dato crudo en solvento-data-prod.shinkansen.shinkansen_treasury. Esa misma tabla es la que el Cloud Run lee en el paso 2 para calcular la siguiente extracción incremental.

Nota: la primera carga debe hacerse manual (si la tabla está vacía, MAX(inserted_at) es nulo y la extracción incremental aborta con un 400). No hay lógica de negocio en la ingesta.

Tablas que alimenta

Tabla	Contenido	Job que la carga
`solvento-data-prod.shinkansen.shinkansen_treasury`	Payout orders crudos de la Treasury API de Shinkansen	Dataflow `treasury-shinkansen-api-prod-20260120-110700`

Consumo aguas abajo

Metabase / análisis de tesorería leen directamente de solvento-data-prod.shinkansen.shinkansen_treasury.
Si existe una capa de staging sobre esta tabla, vive en dbt-analytics (models/staging/shinkansen — (confirmar)). Por criticidad media no se rastrean los modelos aquí.

Si falla 🛠️

Impacto: la tabla shinkansen.shinkansen_treasury deja de recibir payout orders nuevos. Como la extracción es incremental con lookback, una corrida fallida suele recuperarse en la siguiente (re-pide desde la última fecha menos el margen).

Síntoma	Dónde mirar	Acción
La tabla no se actualizó hoy	Cloud Scheduler `shinkansen-treasury-api-daily` (¿disparó?) y logs del Cloud Run	Relanzar manualmente el endpoint `/extract-incremental/treasury`; revisar el scheduler
Error 400 "No data found in BigQuery"	Logs Cloud Run (`execute_incremental_extraction`)	La tabla está vacía → hacer la primera carga manual y luego deja correr el incremental
Errores de API (403/429/5xx) o timeouts	Logs Cloud Run + topic `treasury-api-shinkansen-errors` + Cloud Tasks `treasury-api-shinkansen-queue`	Verificar `API_KEY` (header `Shinkansen-Treasury-Api-Key`); los 429/5xx reintentan solos vía Cloud Tasks
Llega al topic pero la RAW no avanza	Dataflow `treasury-shinkansen-api-prod-…`: watermark; backlog de la sub `treasury-api-shinkansen-success-sub`	Transitorios se recuperan solos consumiendo la misma suscripción; si hay duda, relanzar/escalar
Datos en BQ pero "raros" en Metabase	Capa dbt en `dbt-analytics` (si aplica)	Revisar la transformación en el repo, no la ingesta

Escalación: lógica SQL / Metabase → Diego · resto (Scheduler, Cloud Run, API, Dataflow, Pub/Sub) → Martin.

Enlaces

🔗 Vista que unifica los payouts (este flujo + metadata + results): Shinkansen · Payouts unificado. Treasury aporta las columnas que no llegan por webhook (amount, description, recipient_account, origin_account).
📊 Estado en vivo: dashboard "Data Processes" (Cloud Monitoring)
📦 Handler API: api_shinkansen_handler (cloud_run/main.py) · template Dataflow en templates-dataflow → carpeta shinkansen-treasury-api

Última actualización: 2026-06-21 · Si cambia el flujo, el endpoint de la API o el esquema, actualizar esta ficha en el mismo MR.