VectorFin Apache Iceberg Catalog

# From app.vectorfinancials.com/dashboard/data-access:
#   Catalog URI:  https://catalog.vectorfinancials.com/api/catalog
#   Warehouse:    vectorfin
#   Client ID:    <shown in dashboard>
#   Client Secret: <shown ONCE — copy now>
export VF_POLARIS_CLIENT_ID="..."
export VF_POLARIS_CLIENT_SECRET="..."

# pip install "pyiceberg[gcs]" pandas
from pyiceberg.catalog import load_catalog
import os

catalog = load_catalog(
    "vectorfin",
    **{
        "type": "rest",
        "uri": "https://catalog.vectorfinancials.com/api/catalog",
        "warehouse": "vectorfin",
        "credential": f"{os.environ['VF_POLARIS_CLIENT_ID']}:{os.environ['VF_POLARIS_CLIENT_SECRET']}",
        "scope": "PRINCIPAL_ROLE:ALL",
    },
)

print(catalog.list_namespaces(("vectorfin",)))
# [('vectorfin', 'embeddings'), ('vectorfin', 'signals')]

print(catalog.list_tables(("vectorfin", "signals")))
# whystock_score, regime, volatility, sentiment_drift, anomaly

# 1. In your own GCP project, create a service account (or pick an existing one).
# 2. Email the SA email address (e.g. iceberg@your-proj.iam.gserviceaccount.com)
#    to support@vectorfinancials.com — subject "Iceberg GCS grant <org>".
# 3. We run polaris/grant-snowflake-sa.sh equivalent for your SA, scoped to:
#       gs://vectorfinancials-data/warehouse/vectorfin/*
# 4. Authenticate locally as that SA before running pyiceberg scans:
gcloud auth application-default login --impersonate-service-account=iceberg@your-proj.iam.gserviceaccount.com

from pyiceberg.expressions import And, EqualTo, GreaterThanOrEqual

table = catalog.load_table(("vectorfin", "signals", "whystock_score"))

df = table.scan(
    row_filter=And(
        EqualTo("ticker", "AAPL"),
        GreaterThanOrEqual("date", "2024-01-01"),
    ),
    selected_fields=("ticker", "date", "score", "components"),
).to_pandas()

print(df.head())

from pyiceberg.expressions import And, EqualTo, LessThanOrEqual
from datetime import datetime

# What did we know about AAPL signals as of 2024-01-01?
df = table.scan(
    row_filter=And(
        EqualTo("ticker", "AAPL"),
        LessThanOrEqual("knowledge_ts", datetime(2024, 1, 1).isoformat()),
    ),
    selected_fields=("ticker", "date", "score", "effective_ts", "knowledge_ts"),
).to_pandas()

# Latest known row per (ticker, date)
df = (df.sort_values("knowledge_ts")
        .drop_duplicates(["ticker", "date"], keep="last")
        .sort_values("date"))
print(df.tail())