05 — Normalization and SQLite

Duration: 15 min Prerequisites: chapter 04 (ingestion produces a DataFrame).

The contract

After ingestion we have a DataFrame with date, card, description, category, debit, credit. After normalization we want a clean SQL table that’s easy to query:

CREATE TABLE transactions (
    profile     TEXT NOT NULL,
    date        TEXT NOT NULL,    -- ISO 8601 YYYY-MM-DD
    card        TEXT,
    description TEXT NOT NULL,
    category    TEXT NOT NULL,
    amount      REAL NOT NULL     -- signed: + expense, - refund
);

One signed amount column (not two) — much easier for the LLM to write SQL against.

Normalization rules in csv-llm-shared/normalize.py

import pandas as pd

def normalize(df: pd.DataFrame, profile: str = "default") -> pd.DataFrame:
    out = df.copy()

    # 1. Parse dates → ISO YYYY-MM-DD
    out["date"] = pd.to_datetime(out["date"], errors="coerce").dt.strftime("%Y-%m-%d")

    # 2. Compute signed amount: debit is expense (+), credit is refund (-)
    out["amount"] = out["debit"].astype(float) - out["credit"].astype(float)

    # 3. Trim strings
    for col in ("card", "description", "category"):
        out[col] = out[col].astype(str).str.strip()

    # 4. Drop empty rows
    out = out[(out["description"] != "") & (out["date"].notna())]

    # 5. Attach profile
    out["profile"] = profile

    return out[["profile", "date", "card", "description", "category", "amount"]]

Now amount > 0 means money out and amount < 0 means money in.

Why signed amounts

Compare these two SQL queries:

-- Two-column version (debit/credit)
SELECT category, SUM(debit) - SUM(credit) AS net
FROM transactions
GROUP BY category;

-- Signed version (one column)
SELECT category, SUM(amount) AS net
FROM transactions
GROUP BY category;

The LLM writes the second one much more reliably. Less chance of forgetting one of the two columns.

The SQLite layer in csv-llm-shared/db.py

import sqlite3
from pathlib import Path
import pandas as pd

SCHEMA = """
CREATE TABLE IF NOT EXISTS transactions (
    profile     TEXT NOT NULL,
    date        TEXT NOT NULL,
    card        TEXT,
    description TEXT NOT NULL,
    category    TEXT NOT NULL,
    amount      REAL NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_profile_date ON transactions(profile, date);
CREATE INDEX IF NOT EXISTS idx_category    ON transactions(category);
"""


def open_db(path: str) -> sqlite3.Connection:
    Path(path).parent.mkdir(parents=True, exist_ok=True)
    conn = sqlite3.connect(path)
    conn.executescript(SCHEMA)
    return conn


def load_dataframe(conn, df: pd.DataFrame, profile: str) -> int:
    """(Re)load a profile: delete existing rows then insert."""
    conn.execute("DELETE FROM transactions WHERE profile = ?", (profile,))
    df.to_sql("transactions", conn, if_exists="append", index=False)
    conn.commit()
    return len(df)

The SELECT-only safe query

This is the piece the LLM will use as a tool. We must guarantee it cannot mutate data:

import re

FORBIDDEN = re.compile(
    r"\b(INSERT|UPDATE|DELETE|DROP|ALTER|CREATE|REPLACE|ATTACH|DETACH|PRAGMA)\b",
    re.IGNORECASE,
)


def safe_query(conn, sql: str) -> pd.DataFrame:
    """SELECT-only. Raises ValueError on anything else."""
    if not sql.strip().lower().startswith("select"):
        raise ValueError("SQL refused: only SELECT statements allowed.")
    if FORBIDDEN.search(sql):
        raise ValueError(f"SQL refused: forbidden keyword detected.")
    if ";" in sql.rstrip().rstrip(";"):
        raise ValueError("SQL refused: multiple statements not allowed.")
    return pd.read_sql_query(sql, conn)

Three defenses:

1. Must start with SELECT.
2. No forbidden keyword anywhere (including in comments — -- is preserved by sqlite parser).
3. No multi-statement (a single trailing ; is OK, anything else raises).

Defense in depth

The safe_query wrapper is layer 1. Layer 2 is opening the SQLite connection in read-only mode (we’ll add ?mode=ro in chapter 09 / production). Layer 3 is the OS file permissions. Never rely on a single defense.

End-to-end pipeline

import sys
sys.path.insert(0, "csv-llm-shared")
import ingest, normalize, db

# 1. Read the CSV
df = ingest.read_csv("csv-llm-ollama/data/data1-anonymized.csv")

# 2. Normalize
nf = normalize.normalize(df, profile="data1")

# 3. Load into SQLite
conn = db.open_db("csv-llm-ollama/.cache/csv-llm-ollama.sqlite")
inserted = db.load_dataframe(conn, nf, profile="data1")
print(f"{inserted} rows inserted")
# 734 rows inserted

# 4. Run a safe query
result = db.safe_query(conn, """
    SELECT category, SUM(amount) AS total
    FROM transactions
    WHERE profile = 'data1' AND amount > 0
    GROUP BY category
    ORDER BY total DESC
    LIMIT 5
""")
print(result)

Real output:

       category     total
0       Housing  30895.29
1        Travel  16597.44
2      Business  14386.39
3     Groceries   9725.93
4     Education   8987.85

This matches what we computed in chapter 02 — the pipeline is consistent end-to-end.

What about duplicates?

A naive ingestion would insert the same row twice if you reload the CSV. Two strategies:

1. DELETE-then-INSERT by profile (what load_dataframe does above). Simple and good enough for our case.
2. UNIQUE index on (profile, date, description, amount) with INSERT OR IGNORE. More elegant but trickier if a real merchant legitimately charges you twice on the same day.

We pick strategy 1.

Takeaways

• After normalization, we have one signed amount column, making SQL queries simpler for the LLM.
• The DB has one table, two indices, and 734 rows for the demo profile.
• safe_query is the only way the LLM touches the database. SELECT-only, single statement, no keyword bypass.
• Reloading a CSV deletes the previous rows for the same profile — no manual cleanup needed.

Next: Categorization (rules and LLM) →