Recipes¶

These patterns add validation around the intentionally small client API.

Production-oriented value wrapper¶

from datetime import date
import os

import pacifico


def get_snapshot(tickers):
    today = date.today()
    frame = pacifico.request(
        token=os.environ["PACIFICO_API_TOKEN"],
        ticker=list(tickers),
        dateStart=today,
        dateEnd=today,
        timeOut=600,
        timeFrecuency=1.0,
        format="dataFrame",
    )

    if "Ticker" not in frame.columns:
        raise RuntimeError("Unexpected Pacífico value schema.")
    if frame["Ticker"].eq("Error").any():
        raise RuntimeError("Pacífico returned a value error row.")
    return frame

Explicit dates prevent stale import-time defaults. A list preserves the client-supported multi-ticker path.

Metadata-driven selection¶

catalog = pacifico.request(
    "token.key",
    ticker="metadata",
)

eligible = catalog.loc[
    (catalog["Country"] == "Chile")
    & (catalog["Market"] == "RF")
    & (catalog["Family"] == "BCP"),
    "Ticker",
].dropna().drop_duplicates()

values = pacifico.request(
    "token.key",
    ticker=eligible.head(20).tolist(),
)

Cap batch sizes according to the workload rather than sending an unbounded catalog in one request.

Enforce the author client-side¶

Current server paths do not uniformly enforce source/version filters. Validate the result:

requested_author = "Pacifico"
expected_author_label = "Pacifico"

data = pacifico.request(
    "token.key",
    ticker="CHILE",
    author=requested_author,
)

scenario = data["Scenario"].astype(str).str.casefold()
label = expected_author_label.casefold()
allowed = scenario.eq(label) | scenario.str.startswith(label + "-")
unexpected = data[~allowed]
if not unexpected.empty:
    raise RuntimeError("Response contains an unexpected scenario.")

The server presents the public author under the display label Pacifico, normalizing any legacy internal provider identifier. A named version can extend it to Pacifico-<VERSION> (and may add a scenario suffix), so validate either the exact expected scenario or the author-label prefix appropriate to your policy. For reports, apply the same pattern to Author.

Select one field and pivot¶

history = pacifico.request(
    "token.key",
    ticker=["CHILE", "BCP0600323"],
    dateStart=start,
    dateEnd=end,
    fieldType="Price",
)

prices = history.pivot_table(
    index="Date Effective",
    columns="Ticker",
    values="Value",
    aggfunc="last",
)

Choose aggfunc deliberately. Multiple publications, scenarios, fixings, or tenors can otherwise collide at the same effective time.

Validate a historical boundary¶

effective = history["Date Effective"]

if not effective.empty:
    returned_dates = effective.dt.date
    if returned_dates.min() < start or returned_dates.max() > end:
        raise RuntimeError("Response falls outside the requested interval.")

This detects out-of-range output, but not a normalized request that returns a valid subset. Add domain-specific expected-date checks for audit-sensitive pipelines.

Separate report values by type¶

reports = pacifico.request(
    "token.key",
    item="97004000-5",
)

numbers = reports[
    reports["Value Type"].isin(["Integer", "Double"])
].copy()

text = reports[
    reports["Value Type"] == "String"
].copy()

Do not cast the full mixed Value column to a single dtype.

Keep raw payloads during onboarding¶

When adding a new ticker family, report document, or application:

import json

raw = pacifico.request(
    "token.key",
    document="<DOCUMENT>",
    item="<ITEM>",
    format="json",
)

decoded = json.loads(raw)
if not isinstance(decoded, (dict, list)):
    raise RuntimeError("Unexpected top-level JSON type.")

Review the sanitized raw schema, then implement and test DataFrame assumptions.

Inspect an application before running it¶

contract = pacifico.request(
    "token.key",
    app="<APPLICATION_NAME>",
    help=True,
)

types = contract.loc[
    contract["Subsection"] == "Type",
    ["Section", "Value"],
]
print(types)

Only send approved arguments after checking the current contract. Never use a credential file as a generic File argument.

Add bounded retries outside the client¶

import random
import time

import requests


def request_with_retry(request_call, attempts=3):
    for attempt in range(attempts):
        try:
            return request_call()
        except (requests.ConnectionError, requests.Timeout):
            if attempt + 1 == attempts:
                raise
            delay = (2 ** attempt) + random.random()
            time.sleep(delay)

Use this only when duplicate execution is safe. Do not broadly catch every exception: TypeError, ValueError, authentication failures, and schema failures need correction rather than retry.

Export with schema checks¶

required = {
    "Scenario",
    "Date Publication",
    "Date Effective",
    "Ticker",
    "Value",
    "Field",
}

missing = required.difference(data.columns)
if missing:
    raise RuntimeError(f"Missing columns: {sorted(missing)}")

data.to_parquet("outputs/value_snapshot.parquet", index=False)

Parquet export is performed by pandas and requires a compatible optional engine. It is separate from the client's built-in CSV/text writer.