asset-history Chunk 1 — Scaffold + Schema

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: Create the asset-history standalone Python project with a working SQLite schema, full CRUD layer, and passing tests — no network access required.

Architecture: New standalone Python project at ~/projects/asset-history/. Three SQLite tables (tickers, price_sources, prices) managed by db.py. All tests run without network. The DB and log files are gitignored; only source code is committed.

Tech Stack: Python 3.11+, SQLite (stdlib), pytest, python-dotenv, Typer (declared in pyproject.toml but not used until Chunk 2)

Spec: /mnt/d/FSS/KB/Business/_WorkingOn/Projects/asset-history/asset-history-design.md

---

File Map

Action	File	Responsibility
Create	~/projects/asset-history/pyproject.toml	Project config, dependencies, CLI entrypoint
Create	~/projects/asset-history/.gitignore	Exclude data/, .env, __pycache__, .venv
Create	~/projects/asset-history/.env.example	Document all required env vars
Create	~/projects/asset-history/src/asset_history/__init__.py	Package marker
Create	~/projects/asset-history/src/asset_history/db.py	Schema creation, CRUD helpers, seed data
Create	~/projects/asset-history/tests/__init__.py	Test package marker
Create	~/projects/asset-history/tests/test_db.py	Full DB test suite

---

Task 1: Project scaffold

Files:

• Create: ~/projects/asset-history/ (new directory, new git repo)

• Create: pyproject.toml

• Create: .gitignore

• Create: .env.example

• Step 1.1: Create project directory and git repo

mkdir -p ~/projects/asset-history
cd ~/projects/asset-history
git init

• Step 1.2: Create pyproject.toml

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[project]
name = "asset-history"
version = "0.1.0"
requires-python = ">=3.11"
dependencies = [
    "yfinance>=0.2",
    "requests>=2.31",
    "typer>=0.12",
    "python-dotenv>=1.0",
    "pyyaml>=6.0",
]

[project.optional-dependencies]
dev = [
    "pytest>=8.0",
    "pytest-mock>=3.12",
]

[project.scripts]
asset-history = "asset_history.cli:app"

[tool.pytest.ini_options]
testpaths = ["tests"]
markers = ["slow: integration tests requiring network and API keys"]

• Step 1.3: Create .gitignore

data/
.env
.venv/
__pycache__/
*.pyc
*.egg-info/
dist/
.pytest_cache/

• Step 1.4: Create .env.example

# Path to notify_manager utility (FSS shared notification tool)
NOTIFY_MANAGER_PATH=/mnt/d/FSS/Software/Utils/PythonUtils

# Shared email credential used by notify_manager
EMAIL_PASSWORD=

# Tiingo free API key — register at https://www.tiingo.com/
TIINGO_API_KEY=

# FRED free API key — register at https://fred.stlouisfed.org/docs/api/api_key.html
FRED_API_KEY=

• Step 1.5: Create package structure

mkdir -p src/asset_history
mkdir -p tests
mkdir -p data
touch src/asset_history/__init__.py
touch tests/__init__.py

• Step 1.6: Create virtual environment and install

cd ~/projects/asset-history
python3 -m venv .venv
source .venv/bin/activate
pip install -e ".[dev]"

Expected: installs cleanly, no errors.

• Step 1.7: Verify pytest discovers tests (none yet)

pytest

Expected: no tests ran or collected 0 items

• Step 1.8: Commit scaffold

git add pyproject.toml .gitignore .env.example src/ tests/
git commit -m "feat: project scaffold — pyproject.toml, package structure, gitignore"

---

Task 2: Database schema (TDD)

Files:

• Create: src/asset_history/db.py
• Create: tests/test_db.py

Step 2a: Write failing tests first

• Step 2.1: Write tests/test_db.py

"""Tests for SQLite schema creation, CRUD, constraints, and seed data."""
import json
import sqlite3
import tempfile
from datetime import datetime, timezone
from pathlib import Path

import pytest

from asset_history.db import (
    create_tables,
    get_connection,
    insert_ticker,
    insert_price_source,
    upsert_price,
    get_prices,
    seed_tickers,
)


@pytest.fixture
def db_path(tmp_path):
    """Temporary DB path — isolated per test."""
    return tmp_path / "test.db"


@pytest.fixture
def conn(db_path):
    """Open connection with tables created."""
    con = get_connection(db_path)
    create_tables(con)
    yield con
    con.close()


# ---------------------------------------------------------------------------
# Table creation
# ---------------------------------------------------------------------------

class TestCreateTables:
    def test_creates_tickers_table(self, conn):
        row = conn.execute(
            "SELECT name FROM sqlite_master WHERE type='table' AND name='tickers'"
        ).fetchone()
        assert row is not None

    def test_creates_price_sources_table(self, conn):
        row = conn.execute(
            "SELECT name FROM sqlite_master WHERE type='table' AND name='price_sources'"
        ).fetchone()
        assert row is not None

    def test_creates_prices_table(self, conn):
        row = conn.execute(
            "SELECT name FROM sqlite_master WHERE type='table' AND name='prices'"
        ).fetchone()
        assert row is not None

    def test_idempotent(self, conn):
        """create_tables() can be called twice without error."""
        create_tables(conn)  # second call


# ---------------------------------------------------------------------------
# tickers CRUD
# ---------------------------------------------------------------------------

class TestTickersCRUD:
    def test_insert_and_query_ticker(self, conn):
        insert_ticker(conn, ticker="XIC.TO", name="iShares TSX Composite",
                      country="ca", asset_type="etf",
                      sources=["yahoo", "bank_of_canada"])
        row = conn.execute(
            "SELECT ticker, name, country, asset_type, sources FROM tickers WHERE ticker=?",
            ("XIC.TO",)
        ).fetchone()
        assert row is not None
        assert row[0] == "XIC.TO"
        assert row[1] == "iShares TSX Composite"
        assert row[2] == "ca"
        assert row[3] == "etf"
        assert json.loads(row[4]) == ["yahoo", "bank_of_canada"]

    def test_ticker_primary_key_conflict(self, conn):
        insert_ticker(conn, ticker="SPY", name="SPDR S&P 500", country="us",
                      asset_type="etf", sources=["yahoo"])
        with pytest.raises(sqlite3.IntegrityError):
            insert_ticker(conn, ticker="SPY", name="Duplicate", country="us",
                          asset_type="etf", sources=["yahoo"])


# ---------------------------------------------------------------------------
# price_sources CRUD
# ---------------------------------------------------------------------------

class TestPriceSourcesCRUD:
    def test_insert_and_query_price_source(self, conn):
        insert_ticker(conn, "XIC.TO", "iShares TSX", "ca", "etf", ["yahoo"])
        insert_price_source(
            conn,
            ticker="XIC.TO",
            date="2024-01-15",
            source="yahoo",
            close=28.10,
            adj_close=28.10,
            dividend_amount=0.0,
        )
        row = conn.execute(
            "SELECT ticker, date, source, close, adj_close, dividend_amount "
            "FROM price_sources WHERE ticker=? AND date=? AND source=?",
            ("XIC.TO", "2024-01-15", "yahoo")
        ).fetchone()
        assert row is not None
        assert row[2] == "yahoo"
        assert abs(row[3] - 28.10) < 0.001

    def test_price_source_upsert_updates_fetched_at(self, conn):
        """Re-inserting a (ticker, date, source) row updates fetched_at."""
        insert_ticker(conn, "XIC.TO", "iShares TSX", "ca", "etf", ["yahoo"])
        insert_price_source(conn, "XIC.TO", "2024-01-15", "yahoo",
                            close=28.10, adj_close=28.10, dividend_amount=0.0)
        insert_price_source(conn, "XIC.TO", "2024-01-15", "yahoo",
                            close=28.20, adj_close=28.20, dividend_amount=0.0)
        rows = conn.execute(
            "SELECT COUNT(*) FROM price_sources WHERE ticker='XIC.TO' AND date='2024-01-15'"
        ).fetchone()
        assert rows[0] == 1  # upsert, not duplicate

    def test_allows_null_adj_close(self, conn):
        """Government index sources may have no adj_close."""
        insert_ticker(conn, "XIC.TO", "iShares TSX", "ca", "etf", ["bank_of_canada"])
        insert_price_source(conn, "XIC.TO", "2024-01-15", "bank_of_canada",
                            close=21_000.0, adj_close=None, dividend_amount=0.0)
        row = conn.execute(
            "SELECT adj_close FROM price_sources WHERE ticker='XIC.TO' AND date='2024-01-15'"
        ).fetchone()
        assert row[0] is None


# ---------------------------------------------------------------------------
# prices CRUD + confidence constraint
# ---------------------------------------------------------------------------

class TestPricesCRUD:
    def test_insert_and_query_prices(self, conn):
        insert_ticker(conn, "XIC.TO", "iShares TSX", "ca", "etf", ["yahoo"])
        upsert_price(conn, ticker="XIC.TO", date="2024-01-15",
                     close=28.10, adj_close=28.10, dividend_amount=0.0,
                     primary_source="yahoo", confidence="verified")
        rows = get_prices(conn, ticker="XIC.TO", since="2024-01-01")
        assert len(rows) == 1
        assert rows[0]["date"] == "2024-01-15"
        assert rows[0]["confidence"] == "verified"

    def test_confidence_verified_allowed(self, conn):
        insert_ticker(conn, "SPY", "SPDR S&P 500", "us", "etf", ["yahoo"])
        upsert_price(conn, "SPY", "2024-01-15", 480.0, 480.0, 0.0, "yahoo", "verified")

    def test_confidence_single_source_allowed(self, conn):
        insert_ticker(conn, "SPY", "SPDR S&P 500", "us", "etf", ["yahoo"])
        upsert_price(conn, "SPY", "2024-01-15", 480.0, 480.0, 0.0, "yahoo", "single-source")

    def test_confidence_conflict_allowed(self, conn):
        insert_ticker(conn, "SPY", "SPDR S&P 500", "us", "etf", ["yahoo"])
        upsert_price(conn, "SPY", "2024-01-15", 480.0, 480.0, 0.0, "yahoo", "conflict")

    def test_confidence_invalid_rejected(self, conn):
        insert_ticker(conn, "SPY", "SPDR S&P 500", "us", "etf", ["yahoo"])
        with pytest.raises(sqlite3.IntegrityError):
            upsert_price(conn, "SPY", "2024-01-15", 480.0, 480.0, 0.0, "yahoo", "INVALID")

    def test_get_prices_excludes_conflicts_by_default(self, conn):
        insert_ticker(conn, "SPY", "SPDR S&P 500", "us", "etf", ["yahoo"])
        upsert_price(conn, "SPY", "2024-01-15", 480.0, 480.0, 0.0, "yahoo", "verified")
        upsert_price(conn, "SPY", "2024-01-16", 481.0, 481.0, 0.0, "yahoo", "conflict")
        rows = get_prices(conn, "SPY", since="2024-01-01")
        assert len(rows) == 1
        assert rows[0]["date"] == "2024-01-15"

    def test_get_prices_include_conflicts_flag(self, conn):
        insert_ticker(conn, "SPY", "SPDR S&P 500", "us", "etf", ["yahoo"])
        upsert_price(conn, "SPY", "2024-01-15", 480.0, 480.0, 0.0, "yahoo", "verified")
        upsert_price(conn, "SPY", "2024-01-16", 481.0, 481.0, 0.0, "yahoo", "conflict")
        rows = get_prices(conn, "SPY", since="2024-01-01", include_conflicts=True)
        assert len(rows) == 2


# ---------------------------------------------------------------------------
# seed_tickers
# ---------------------------------------------------------------------------

class TestSeedTickers:
    def test_seed_populates_xic(self, conn):
        seed_tickers(conn)
        row = conn.execute(
            "SELECT ticker, country, sources FROM tickers WHERE ticker='XIC.TO'"
        ).fetchone()
        assert row is not None
        assert row[1] == "ca"
        assert "yahoo" in json.loads(row[2])
        assert "bank_of_canada" in json.loads(row[2])

    def test_seed_populates_spy(self, conn):
        seed_tickers(conn)
        row = conn.execute(
            "SELECT ticker, country, sources FROM tickers WHERE ticker='SPY'"
        ).fetchone()
        assert row is not None
        assert row[1] == "us"
        sources = json.loads(row[2])
        assert "yahoo" in sources
        assert "tiingo" in sources
        assert "fred" in sources

    def test_seed_idempotent(self, conn):
        """Calling seed_tickers() twice should not raise or duplicate."""
        seed_tickers(conn)
        seed_tickers(conn)
        count = conn.execute("SELECT COUNT(*) FROM tickers").fetchone()[0]
        assert count == 2

• Step 2.2: Run tests — verify all fail with ImportError

cd ~/projects/asset-history
source .venv/bin/activate
pytest tests/test_db.py -v

Expected: All tests FAIL with ImportError: cannot import name 'create_tables' from 'asset_history.db'

Step 2b: Implement db.py

• Step 2.3: Create src/asset_history/db.py

"""
SQLite schema, CRUD helpers, and seed data for asset-history.

Three tables:
  tickers       — registered assets (primary key: ticker)
  price_sources — raw EOD data per source, append-only audit trail
  prices        — reconciled SSoT, one row per (ticker, date)
"""
import json
import sqlite3
from datetime import datetime, timezone
from pathlib import Path
from typing import Any


def get_connection(db_path: Path | str) -> sqlite3.Connection:
    """Open a SQLite connection with foreign keys and WAL mode enabled."""
    conn = sqlite3.connect(str(db_path))
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA journal_mode=WAL")
    conn.execute("PRAGMA foreign_keys=ON")
    return conn


def create_tables(conn: sqlite3.Connection) -> None:
    """Create all tables if they don't already exist. Idempotent."""
    conn.executescript("""
        CREATE TABLE IF NOT EXISTS tickers (
            ticker      TEXT PRIMARY KEY,
            name        TEXT,
            country     TEXT,
            asset_type  TEXT,
            sources     TEXT,
            added_at    TIMESTAMP NOT NULL
        );

        CREATE TABLE IF NOT EXISTS price_sources (
            ticker          TEXT NOT NULL,
            date            DATE NOT NULL,
            source          TEXT NOT NULL,
            close           REAL,
            adj_close       REAL,
            dividend_amount REAL DEFAULT 0.0,
            fetched_at      TIMESTAMP NOT NULL,
            PRIMARY KEY (ticker, date, source)
        );

        CREATE TABLE IF NOT EXISTS prices (
            ticker          TEXT NOT NULL,
            date            DATE NOT NULL,
            close           REAL NOT NULL,
            adj_close       REAL NOT NULL,
            dividend_amount REAL NOT NULL DEFAULT 0.0,
            primary_source  TEXT NOT NULL,
            confidence      TEXT NOT NULL
                CHECK (confidence IN ('verified', 'single-source', 'conflict')),
            PRIMARY KEY (ticker, date)
        );
    """)
    conn.commit()


def insert_ticker(
    conn: sqlite3.Connection,
    ticker: str,
    name: str,
    country: str,
    asset_type: str,
    sources: list[str],
) -> None:
    """Insert a new ticker. Raises sqlite3.IntegrityError on duplicate."""
    conn.execute(
        "INSERT INTO tickers (ticker, name, country, asset_type, sources, added_at) "
        "VALUES (?, ?, ?, ?, ?, ?)",
        (ticker, name, country, asset_type, json.dumps(sources), _now()),
    )
    conn.commit()


def insert_price_source(
    conn: sqlite3.Connection,
    ticker: str,
    date: str,
    source: str,
    close: float | None,
    adj_close: float | None,
    dividend_amount: float,
) -> None:
    """Insert or replace a price_sources row (upsert on primary key conflict)."""
    conn.execute(
        "INSERT OR REPLACE INTO price_sources "
        "(ticker, date, source, close, adj_close, dividend_amount, fetched_at) "
        "VALUES (?, ?, ?, ?, ?, ?, ?)",
        (ticker, date, source, close, adj_close, dividend_amount, _now()),
    )
    conn.commit()


def upsert_price(
    conn: sqlite3.Connection,
    ticker: str,
    date: str,
    close: float,
    adj_close: float,
    dividend_amount: float,
    primary_source: str,
    confidence: str,
) -> None:
    """Insert or replace a reconciled prices row."""
    conn.execute(
        "INSERT OR REPLACE INTO prices "
        "(ticker, date, close, adj_close, dividend_amount, primary_source, confidence) "
        "VALUES (?, ?, ?, ?, ?, ?, ?)",
        (ticker, date, close, adj_close, dividend_amount, primary_source, confidence),
    )
    conn.commit()


def get_prices(
    conn: sqlite3.Connection,
    ticker: str,
    since: str,
    include_conflicts: bool = False,
) -> list[dict[str, Any]]:
    """
    Return reconciled price rows for ticker since (inclusive) a date string.
    Excludes conflict rows by default.
    """
    if include_conflicts:
        rows = conn.execute(
            "SELECT * FROM prices WHERE ticker=? AND date>=? ORDER BY date ASC",
            (ticker, since),
        ).fetchall()
    else:
        rows = conn.execute(
            "SELECT * FROM prices WHERE ticker=? AND date>=? AND confidence!='conflict' "
            "ORDER BY date ASC",
            (ticker, since),
        ).fetchall()
    return [dict(r) for r in rows]


def seed_tickers(conn: sqlite3.Connection) -> None:
    """
    Pre-populate the tickers table with the initial tracked assets.
    Idempotent — uses INSERT OR IGNORE.
    """
    seeds = [
        ("XIC.TO", "iShares Core S&P/TSX Composite ETF", "ca", "etf",
         ["yahoo", "bank_of_canada"]),
        ("SPY", "SPDR S&P 500 ETF Trust", "us", "etf",
         ["yahoo", "tiingo", "fred"]),
    ]
    for ticker, name, country, asset_type, sources in seeds:
        conn.execute(
            "INSERT OR IGNORE INTO tickers "
            "(ticker, name, country, asset_type, sources, added_at) VALUES (?, ?, ?, ?, ?, ?)",
            (ticker, name, country, asset_type, json.dumps(sources), _now()),
        )
    conn.commit()


def _now() -> str:
    return datetime.now(timezone.utc).isoformat()

• Step 2.4: Run all tests — verify they pass

pytest tests/test_db.py -v

Expected: All tests PASS. Output shows test count matching the test file.

• Step 2.5: Commit

git add src/asset_history/db.py tests/test_db.py
git commit -m "feat: SQLite schema, CRUD helpers, and seed data — all tests pass"

---

Task 3: Final verification

• Step 3.1: Run full test suite with no network

pytest -v

Expected: All tests pass. No network calls made.

• Step 3.2: Verify data directory is gitignored

touch data/test.db
git status

Expected: data/test.db does NOT appear in git status output (it is ignored).

• Step 3.3: Verify package is importable

python3 -c "from asset_history.db import create_tables, seed_tickers; print('OK')"

Expected: OK

• Step 3.4: Final commit (if any cleanup needed)

git status
# If clean, no commit needed. If any stray files, clean them up and commit.

---

Chunk 1 Gate

All of the following must be true before starting Chunk 2:

• pytest tests/test_db.py -v — all tests pass
• git log --oneline shows at least 2 commits
• data/ directory is gitignored
• python3 -c "from asset_history.db import create_tables; print('OK')" prints OK

---

Completion Record — 2026-03-22

Environment note

python3 -m venv .venv failed (missing python3.12-venv package, same environment constraint as monorepo). Resolution: installed all dependencies system-wide via pip3 install --break-system-packages -e ".[dev]". All tests run against the system Python 3.12.3. Chunk 2 implementer should be aware of this constraint.

Pytest output

============================= test session starts ==============================
platform linux -- Python 3.12.3, pytest-9.0.2, pluggy-1.6.0
rootdir: /home/ta/projects/asset-history
configfile: pyproject.toml
plugins: anyio-4.12.1, mock-3.15.1
collected 19 items

tests/test_db.py::TestCreateTables::test_creates_tickers_table PASSED
tests/test_db.py::TestCreateTables::test_creates_price_sources_table PASSED
tests/test_db.py::TestCreateTables::test_creates_prices_table PASSED
tests/test_db.py::TestCreateTables::test_idempotent PASSED
tests/test_db.py::TestTickersCRUD::test_insert_and_query_ticker PASSED
tests/test_db.py::TestTickersCRUD::test_ticker_primary_key_conflict PASSED
tests/test_db.py::TestPriceSourcesCRUD::test_insert_and_query_price_source PASSED
tests/test_db.py::TestPriceSourcesCRUD::test_price_source_upsert_updates_fetched_at PASSED
tests/test_db.py::TestPriceSourcesCRUD::test_allows_null_adj_close PASSED
tests/test_db.py::TestPricesCRUD::test_insert_and_query_prices PASSED
tests/test_db.py::TestPricesCRUD::test_confidence_verified_allowed PASSED
tests/test_db.py::TestPricesCRUD::test_confidence_single_source_allowed PASSED
tests/test_db.py::TestPricesCRUD::test_confidence_conflict_allowed PASSED
tests/test_db.py::TestPricesCRUD::test_confidence_invalid_rejected PASSED
tests/test_db.py::TestPricesCRUD::test_get_prices_excludes_conflicts_by_default PASSED
tests/test_db.py::TestPricesCRUD::test_get_prices_include_conflicts_flag PASSED
tests/test_db.py::TestSeedTickers::test_seed_populates_xic PASSED
tests/test_db.py::TestSeedTickers::test_seed_populates_spy PASSED
tests/test_db.py::TestSeedTickers::test_seed_idempotent PASSED

19 passed in 0.88s

Git log

9dc9354 feat: SQLite schema, CRUD helpers, and seed data — all tests pass
cd0501f feat: project scaffold — pyproject.toml, package structure, gitignore

Gate verification

Check	Result
pytest tests/test_db.py -v — all pass	✓ 19/19
git log --oneline shows ≥ 2 commits	✓ 2 commits
data/test.db absent from git status	✓ nothing to commit, working tree clean
from asset_history.db import create_tables; print('OK')	✓ OK

Files created

File	Description
~/projects/asset-history/pyproject.toml	Hatchling build, all declared deps, pytest config
~/projects/asset-history/.gitignore	Excludes data/, .env, .venv/, caches
~/projects/asset-history/.env.example	Documents all 4 env vars
~/projects/asset-history/src/asset_history/__init__.py	Package marker
~/projects/asset-history/src/asset_history/db.py	Full schema + CRUD + seed
~/projects/asset-history/tests/__init__.py	Test package marker
~/projects/asset-history/tests/test_db.py	19 tests across 4 test classes