Custom AI Models

Yeriko Vargas

AI Engineer · AI Systems Architect · Detroit, MI

Code architecture

How I structure code so a system still reads clearly after its fiftieth change. Each pattern is written down once, enforced on every build, and published here with the decisions behind it.

Pattern 01The builder

Stage contract · selector-first ingestion · three data tiers

in use
Pattern 02The agent rules

AGENTS.md · one entry file · functions only · ask before any env

in use
Decision recordsADR log

Context, decision and consequences for each architecture call

publishing next

I don't just print code

Seven years of my own Python functions, built for the way I work.

Every function I reuse was written, broken and rewritten against real workflows. I know where each one sits, what it touches and what it returns, so when something fails I know which stage to open, and when something is slow I know which function to tune.

Pattern 01 · The builder

One folder. One command. The whole pipeline.

A builder is a self-contained mini-app: python run.py runs the full pipeline. It lives inside a project but runs on its own, and a shared selector at the project root feeds every builder the same way.

00select

rows in: collect, filter, scan or one path

01build

validate · normalize

02process

tag · classify

03extract

→ data/processed

04analyze

→ data/exports

$ /builder_str <name> <domain>

The shape

project treewhat the scaffold creates16 lines
<project_root>/
├── selector_df_rows.py          ← create only if it doesn't exist yet
│
└── <name>/                      ← the builder folder
    ├── run.py                   ← entry point; selector is ALWAYS first call
    ├── src_1/
    │   ├── __init__.py
    │   ├── _1_build_<domain>.py ← validate/normalize df received from selector
    │   ├── _2_proc_<domain>.py  ← tag / classify
    │   ├── _3_ext_<domain>.py   ← save to data/processed/
    │   ├── _4_ana_<domain>.py   ← analyze + save to data/exports/
    │   └── _5_util_<domain>.py  ← shared helpers; only file other stages import
    └── data/
        ├── raw/.gitkeep
        ├── processed/.gitkeep
        └── exports/.gitkeep

No config/ at scaffold time. CFG lives inline in run.py until the project grows.

The stage contract · enforced always

every stageone signature3 lines
def fn(df, paths, cfg):   # receives df, returns df
    ...
    return df

The source

The full implementation, exactly as the scaffold writes it. Open any file.

run.pyentry point · selector always first42 lines
import os
import sys
 
_HERE = os.path.dirname(os.path.abspath(__file__))
_ROOT = os.path.dirname(_HERE)
 
sys.path.insert(0, _ROOT)   # selector_df_rows lives here
sys.path.insert(0, _HERE)   # src_1 lives here
 
import selector_df_rows as selector
 
from src_1._1_build_<domain> import build
from src_1._2_proc_<domain>  import process
from src_1._3_ext_<domain>   import extract
from src_1._4_ana_<domain>   import analyze
 
CFG = {
    # numbers, flags, names only — no paths
}
 
PATHS = {
    "raw":       os.path.join(_HERE, "data", "raw"),
    "processed": os.path.join(_HERE, "data", "processed"),
    "exports":   os.path.join(_HERE, "data", "exports"),
}
 
 
def run():
    print(f"\n  ┌─ <name> {'─' * (40 - len('<name>'))}┐")
 
    df = selector.select()        # always first
    df = build(df, PATHS, CFG)
    df = process(df, PATHS, CFG)
    df = extract(df, PATHS, CFG)
    df = analyze(df, PATHS, CFG)
 
    print("  └──────────────────────────────────────────┘\n")
    return df
 
 
if __name__ == "__main__":
    run()
selector_df_rows.pyproject root · shared by every builder86 lines
import os
import pandas as pd
 
_HERE = os.path.dirname(os.path.abspath(__file__))
_CSV  = os.path.join(_HERE, "_df_collect.csv")
 
 
def select():
    """Returns a DataFrame. Always the first call in any builder run.py."""
    print("\n  [selector]  use rows?  [Enter = full collect   s = select]  ", end="", flush=True)
    choice = input().strip().lower()
 
    if choice == "s":
        return _select_mode()
    return _default()
 
 
def _default():
    if os.path.exists(_CSV):
        df = pd.read_csv(_CSV)
        print(f"  [selector]  →  {len(df)} rows  (_df_collect.csv)")
        return df
    print("  [selector]  no _df_collect.csv found — returning empty df")
    return pd.DataFrame()
 
 
def _select_mode():
    if os.path.exists(_CSV):
        return _filter_collect()
    return _from_input()
 
 
def _filter_collect():
    df = pd.read_csv(_CSV)
    print(f"\n  collect:  {len(df)} rows   cols: {list(df.columns)}")
 
    if "extension" in df.columns:
        counts = df["extension"].value_counts()
        print("\n  extensions:")
        for ext, n in counts.items():
            print(f"    {ext:<12} {n}")
        val = input("\n  filter by extension  (comma-sep, or Enter to skip): ").strip()
        if val:
            wanted = [e.strip() for e in val.split(",")]
            df = df[df["extension"].isin(wanted)].reset_index(drop=True)
 
    print(f"  [selector]  →  {len(df)} rows  (filtered)")
    return df
 
 
def _from_input():
    print("\n  no _df_collect.csv")
    print("  [f] scan folder    [p] single path    [Enter] skip")
    choice = input("  → ").strip().lower()
 
    if choice == "f":
        folder = input("  folder path: ").strip()
        return _walk(folder)
    if choice == "p":
        path = input("  file path: ").strip()
        name = os.path.basename(path)
        _, ext = os.path.splitext(name)
        return pd.DataFrame([{
            "file_name":  name,
            "file_path":  path,
            "extension":  ext.lower(),
            "size_mb":    round(os.path.getsize(path) / 1e6, 3) if os.path.exists(path) else 0,
        }])
    return pd.DataFrame()
 
 
def _walk(folder):
    rows = []
    for root, _, files in os.walk(folder):
        for f in files:
            full = os.path.join(root, f)
            _, ext = os.path.splitext(f)
            rows.append({
                "file_name":  f,
                "file_path":  full,
                "extension":  ext.lower(),
                "size_mb":    round(os.path.getsize(full) / 1e6, 3),
            })
    df = pd.DataFrame(rows)
    print(f"  [selector]  →  {len(df)} rows  (scanned folder)")
    return df
_1_build_<domain>.pyvalidate · normalize25 lines
import random
import pandas as pd
from src_1._5_util_<domain> import hr
 
 
def build(df, paths, cfg):
    hr("BUILD")
    if df.empty:
        print("   empty df from selector — generating synthetic data")
        df = _synth(cfg)
    else:
        print(f"   df in:  {len(df)} rows  {list(df.columns)}")
    return _validate(df, cfg)
 
 
def _synth(cfg):
    # domain-specific synthetic data for standalone testing
    return pd.DataFrame()
 
 
def _validate(df, cfg):
    before = len(df)
    # domain-specific validation here
    print(f"   validate  {before} → {len(df)} rows")
    return df.reset_index(drop=True)
_2_proc_<domain>.pytag · classify9 lines
from src_1._5_util_<domain> import hr
 
 
def process(df, paths, cfg):
    hr("PROCESS")
    df = df.copy()
    # domain-specific tagging
    print(f"   tagged  {len(df)} rows")
    return df
_3_ext_<domain>.pysave → data/processed9 lines
import os
from src_1._5_util_<domain> import hr, save_csv
 
 
def extract(df, paths, cfg):
    hr("EXTRACT")
    out = os.path.join(paths["processed"], "extracted.csv")
    save_csv(df, out)
    return df
_4_ana_<domain>.pyanalyze → data/exports10 lines
import os
from src_1._5_util_<domain> import hr, save_csv
 
 
def analyze(df, paths, cfg):
    hr("ANALYZE")
    # domain-specific analysis / prints
    out = os.path.join(paths["exports"], f"{cfg['export_name']}.csv")
    save_csv(df, out)
    return df
_5_util_<domain>.pyhelpers · the only cross-import22 lines
import os
import pandas as pd
 
 
def load_csv(path):
    if not os.path.exists(path):
        return None
    return pd.read_csv(path)
 
 
def save_csv(df, path):
    os.makedirs(os.path.dirname(path), exist_ok=True)
    df.to_csv(path, index=False)
    print(f"   saved  {os.path.basename(path)}  ({len(df)} rows)")
 
 
def hr(label=""):
    if label:
        pad = (44 - len(label) - 2) // 2
        print(f"\n{'─'*pad} {label} {'─'*(44-pad-len(label)-2)}")
    else:
        print("─" * 44)

Written as a Claude Code command, so every builder an agent scaffolds follows the same contract.

Pattern 02 · The agent rules

The rules an agent reads before it writes a line.

When an AI agent codes for me, it works inside my architecture, not its own. These rules came out of a multi-builder finance app: auto.py finds every run.py, one file holds every input, and nothing gets installed without a yes.

R1_u_entries.py

the only file a human edits · no logic

R2_f_fns*.py

functions only · one domain per file

R4ask first

never creates an env · never installs

autoauto.py

finds every run.py · stops at the first failure

lawfinance

no look-ahead · time splits · a baseline first

my rulesverbatim · as I wrote them for Cursor46 lines
RULE 1 — USER ENTRY FILE
 
Every project MUST include a file named: _u_entries.py
 
Purpose:
- This is the ONLY file the user edits manually
 
Contains:
- input paths
- tickers / parameters
- runtime options
 
Rules:
- No logic allowed
- Only variables / configs
- All other files MUST import from this file
 
RULE 2 — FUNCTION FILES
 
All core logic lives in: _f_fns*.py
 
Structure:
- _f_fns.py → general functions
- _f_fns_<keyword>.py → grouped by purpose
 
Rules:
- Functions MUST have explicit, readable names
- Each file = one cohesive domain
- Each file handles its own imports
- No execution code (functions only)
 
RULE 4 — ENVIRONMENT CONTROL
 
Before running/installing anything:
 
- ALWAYS ask:
    "Do you want to create a new environment?"
 
If NO:
- Ask:
    "Which environment should be used?"
 
Rules:
- NEVER auto-create environments
- NEVER install packages without confirmation
- ALWAYS respect user's existing setup
AGENTS.mdthe full contract an agent follows · tested210 lines
# AGENTS.md · builders
 
Read this before you write a line. Every agent that works here (Claude Code, Cursor, Codex) follows it.
§1 is the owner's rules, word for word. Everything after expands them and never overrides them.
When a request conflicts with this file, stop and ask.
 
## 0. The one-minute version
 
- `_u_entries.py` is the only file a human edits. You read it. You never put logic in it.
- Logic lives in function files: functions only, nothing runs on import.
- Every builder has one front door, `run.py`. `auto.py` finds every `run.py` and runs them in order.
- Never create an environment or install a package without asking first.
- Data moves one way: raw → processed → exports. Nothing writes back into raw.
- Done means `python auto.py --check` passes and you can name the function you changed and why.
 
## 1. The owner's rules (verbatim)
 
RULE 1 — USER ENTRY FILE
Every project MUST include a file named: _u_entries.py
- This is the ONLY file the user edits manually
- Contains: input paths · tickers / parameters · runtime options
- No logic allowed · Only variables / configs · All other files MUST import from this file
 
RULE 2 — FUNCTION FILES
All core logic lives in: _f_fns*.py
- _f_fns.py → general functions · _f_fns_<keyword>.py → grouped by purpose
- Functions MUST have explicit, readable names · Each file = one cohesive domain
- Each file handles its own imports · No execution code (functions only)
 
RULE 3 — not written yet. Do not invent it. Ask.
 
RULE 4 — ENVIRONMENT CONTROL
Before running/installing anything, ALWAYS ask: "Do you want to create a new environment?"
If NO, ask: "Which environment should be used?"
- NEVER auto-create environments · NEVER install packages without confirmation
- ALWAYS respect user's existing setup
 
## 2. The map
 
```
app/
├── _u_entries.py            ← RULE 1: the only file a human edits
├── AGENTS.md                ← this file
└── builders/
    ├── auto.py              ← finds builders/*/run.py, runs them in number order
    ├── builder_1/           ← get + clean
    ├── builder_2/           ← features + models
    └── builder_3/           ← evaluation + dashboard
        ├── run.py           ← the builder's front door
        ├── README.md        ← what it reads, what it writes, how to run it
        ├── requirements.txt
        ├── .gitignore
        ├── config/config.py ← derived constants only, built from _u_entries
        ├── src/
        │   ├── _0_fns_get/    fn_get_data.py
        │   ├── _1_fns_clean/  fn_clean_data.py
        │   ├── _2_fns_feat/   fn_features.py
        │   ├── _3_fns_model/  fn_model.py
        │   ├── _4_fns_eval/   fn_eval.py
        │   └── _5_fns_utils/  fn_utils.py
        ├── data/
        │   ├── raw/           ← original pulls, never edited
        │   ├── processed/     ← cleaned, aligned, feature tables
        │   └── exports/       ← tables, reports, figures
        └── notebooks/main.ipynb ← playground, never imported by code
```
 
The `src/_N_fns_*/fn_*.py` files are RULE 2's function files: one folder per stage, one domain per file.
A file that serves the whole app stays at the top as `_f_fns.py` or `_f_fns_<keyword>.py`.
 
## 3. `_u_entries.py`: what goes in, what never does
 
```python
# _u_entries.py  ·  edit this file, nothing else
TICKERS     = ["AAPL", "NVDA", "ASML"]
START, END  = "2018-01-01", "2026-06-30"
HORIZON     = 5            # trading days ahead
SEED        = 7
INPUT_DIR   = "data_in"    # relative to the app root
RUN         = {"builder_1": True, "builder_2": True, "builder_3": False}
```
 
- Allowed: literals, lists, dicts, dates as strings, booleans.
- Never allowed: imports, functions, loops, file reads, `if`, computed paths.
- A path is relative to the app root. An absolute path may live here and nowhere else.
- A new parameter gets a one-line comment saying what it changes.
 
## 4. Function files: how a function is written
 
- One job per function, named for what it returns: `get_prices_GET_df`, `clean_gaps_ffill_GET_df`.
- A stage function takes `(df, paths, cfg)` and returns `df`. Paths and cfg come in. They are never built inside.
- Each file imports what it uses at its own top. Files in `_0` to `_4` never import each other. Only `_5_fns_utils` is shared.
- Nothing runs at import time: no top-level calls, no prints, no reads.
- A function that writes a file says so in its name (`_save_`, `_export_`) and returns the path it wrote.
- Raise errors with the stage and the reason (`ValueError("clean: 3 tickers have no rows after 2024-01-02")`). Never pass on them silently.
 
## 5. `run.py`: one builder, one front door
 
```python
import sys
from pathlib import Path
 
HERE = Path(__file__).resolve().parent
ROOT = HERE.parents[1]                       # app/
sys.path[:0] = [str(ROOT), str(HERE)]
 
import _u_entries as U                        # RULE 1: everything starts here
from src._0_fns_get.fn_get_data import get_prices_GET_df
from src._1_fns_clean.fn_clean_data import clean_gaps_ffill_GET_df
 
PATHS = {k: HERE / "data" / k for k in ("raw", "processed", "exports")}
CFG = {"tickers": U.TICKERS, "start": U.START, "end": U.END, "seed": U.SEED}
 
 
def main(check=False):
    df = get_prices_GET_df(None, PATHS, CFG | {"check": check})
    df = clean_gaps_ffill_GET_df(df, PATHS, CFG)
    print(f"✓ builder_1 · {len(df)} rows")
    return 0
 
 
if __name__ == "__main__":
    raise SystemExit(main(check="--check" in sys.argv))
```
 
- `run.py` composes the stages and does nothing else. Logic lives in function files.
- `--check` runs the whole flow on a small, cached slice with no network, no training and no writes outside `data/`.
- The exit code is the truth: 0 means it worked, anything else means it failed.
 
## 6. `auto.py`: finds every `run.py`
 
```python
"""auto.py · runs builders/*/run.py in number order. `python auto.py [--check] [--only 2]`"""
import re, subprocess, sys
from pathlib import Path
 
HERE = Path(__file__).resolve().parent
ROOT = HERE.parent
sys.path.insert(0, str(ROOT))
import _u_entries as U
 
def builders():
    found = [p for p in HERE.glob("builder_*/run.py")]
    return sorted(found, key=lambda p: int(re.search(r"\d+", p.parent.name).group()))
 
def main(argv):
    only = argv[argv.index("--only") + 1] if "--only" in argv else None
    for run in builders():
        name = run.parent.name
        if only and name != f"builder_{only}":
            continue
        if not U.RUN.get(name, True):
            print(f"·  {name}  off in _u_entries.RUN"); continue
        print(f"▶  {name}")
        code = subprocess.call([sys.executable, "run.py", *(["--check"] if "--check" in argv else [])], cwd=run.parent)
        if code:
            print(f"✗  {name} exited {code}. Stopping here."); return code
    print("✓  all builders ran"); return 0
 
if __name__ == "__main__":
    raise SystemExit(main(sys.argv[1:]))
```
 
- Each builder runs in its own process from its own folder, so one builder can't break another.
- It stops at the first failure and names it. It never skips ahead.
- A new `builder_N/run.py` is picked up on its own. Nothing is registered by hand.
 
## 7. RULE 4 in practice: environments
 
1. Before any `pip`, `conda` or `uv` command, ask: "Do you want to create a new environment?"
2. If no, ask: "Which environment should be used?" Use that one exactly, and say its name back.
3. A missing package is reported, never installed: `pandas-ta missing in <env>. Add it?`
4. Once a package is approved it goes into `requirements.txt` in the same change.
5. Never touch the base env, global Python, or another project's env.
 
## 8. Finance law: what makes the numbers worth trusting
 
- **No look-ahead.** A feature at day t uses data up to t and nothing after. Targets are shifted forward, features are never shifted back.
- **Fit on train only.** Scalers, PCA and imputers are fitted on the training window and applied to the rest.
- **Time splits, never shuffles.** Walk-forward or expanding windows. Each fold reports its own metrics.
- **Name the target.** Say what is predicted (for example, the log return `HORIZON` days ahead) and keep it fixed across models.
- **A baseline first.** Last value, then a linear or ARIMA model. An LSTM has to beat both before anyone sees it.
- **Report the spread.** Give MAE, RMSE and directional accuracy per fold, plus the interval coverage when a model gives a distribution.
- **Fix the seed.** Seeds come from `_u_entries.SEED`, so a result can be run again and come out the same.
- **Raw is a snapshot.** Pulls land as `data/raw/<source>_<YYYYMMDD>.csv` and are never edited. Cleaning writes to processed.
- **Fill gaps forward.** Use the trading calendar, forward-fill only, and never back-fill across a split.
- **No live orders.** Nothing here places a trade or moves money. Results are research, not advice.
 
## 9. How an agent works a task
 
```
read      AGENTS.md → the builder's README → _u_entries.py
plan      ≤ 5 lines: which stage, which function, what changes
change    one stage at a time, the smallest change that works
prove     python auto.py --check  (or run.py --check for one builder)
report    file · function · why · what the numbers did
```
 
- Don't rename or move files that work. Don't "tidy" code nobody asked about.
- When the README is wrong after your change, fix the README in the same change.
- When you aren't sure, ask. A question costs less than a wrong rewrite.
 
## 10. Done means
 
- [ ] `python auto.py --check` exits 0
- [ ] No logic in `_u_entries.py`, and no absolute path anywhere else
- [ ] No function file runs anything at import
- [ ] Data only moved raw → processed → exports
- [ ] No environment was created and no package installed without a yes
- [ ] README and requirements.txt match the code

AGENTS.md is the file Cursor, Codex and Claude Code read at a project's root, so one contract covers every agent.