LLD: MM-Cron adapter framework and common output schema v1

status: draft (MMDEP Design, 25 Oct). High-risk design written by Claude (allocation §1); Sonnet builds it (queue T0c). sprint: S2 (1–14 Nov), milestone MC-1 (7 Nov) decisions: TD-041, TD-046, TD-047, TD-048, TD-054, TD-056, TD-059, TD-060 overall: mm-cron/overall.md §3–5; phase: mm-cron/phase-1.md §3–5

Context

MM-Cron has to cover AMFI and 51 AMCs by the end of sprint 3 (setup doc §2). The jobs today are separate scripts that combine fetch, parse and load (KN). Failures are silent (augmentation §1), and holdings are linked to schemes by name (OR X2). This LLD fixes one pipeline, one adapter interface and one output schema. With those fixed: - each AMC becomes a cx 1–2 harness item: configuration, a sample file, expected output; - the API's ingest routes (API phase 1 §4) can validate against the same schema.

It follows ADR-0016 (design docs), ADR-0009 (public data, so free models may build adapters) and the overall design's invariants: no write without checks, full traceability, identity rules.

Decision

1. Where it lives

Sonnet can't reach GitLab yet (setup doc §5), so the framework is built in MMDEP as an open work package at owp/mm-crons/. Its layout is the same as its future home in moneymarche/mm-crons, and the API agent moves it there in one MR after MC-1 (TD-059). Python 3.10+, standard library plus two dependencies: jsonschema (schema validation; the API validates with the same files) and openpyxl (Excel disclosures). PDF parsing (pdfplumber) is added in sprint 3 with the first factsheet adapters, not now.

owp/mm-crons/
  mmcron/
    schema/v1/*.schema.json      one JSON Schema per record type, plus envelope.schema.json and run.schema.json
    schema/__init__.py           load(type) -> schema; validate(type, record) -> list[str]
    adapters/base.py             Adapter, FingerprintResult, ParseContext
    adapters/families/           one module per format family (e.g. cams_excel.py, kfin_excel.py, amfi_text.py)
    registry.py                  loads registry/*.yaml, checks it
    fetch.py                     fetch(url) -> RawFile (bytes, url, status, retrieved_at, sha256)
    archive.py                   store(raw) -> path; manifest line
    validate.py                  shared rules (§5)
    mapping.py                   portfolio name -> scheme codes (§4)
    publish.py                   targets: file | staging | production (§6)
    run.py                       the pipeline (§2) and the run record
    __main__.py                  CLI (§7)
  registry/
    jobs.yaml                    one row per job
    amcs.yaml                    51 rows (§3)
    portfolio_map.yaml           confirmed name -> scheme mappings (§4)
  tests/
    fixtures/<amc_code>/<doc_type>/<yyyy-mm>/{input.<ext>, expected.json}
    test_*.py

2. The pipeline (one function, every job)

run(job_id, env, file=None):

1. Registry: read the job row (source, adapter, family config, target type).
2. Fetch: download from the registry URL, or use --file (a hand-saved sample, or a file whose download was refused). Record retrieved_at, URL, HTTP status, size, sha256.
3. Archive: store the raw file and append a manifest line. This happens before parsing, so a crash still leaves evidence.
4. Fingerprint: adapter.fingerprint(raw). On a mismatch: status stopped-format, a diff of expected against found in the run record, no parse output published, exit code 3.
5. Parse: adapter.parse(raw, ctx) returns envelope records (§3), with parsed_at set.
6. Map: for holdings and fund details, resolve portfolio_name to scheme codes (§4). Records without a confirmed mapping are written to the review list and are not published.
7. Validate: schema, then the shared rules (§5). Any error rule failing for a record set stops the publish (stopped-validation, exit 4). Warnings are recorded and don't stop.
8. Publish: to the target (§6). Record the API's acknowledgement counts and published_at. The API records persisted_at.
9. Run record: one JSON line in logs/runs/<yyyy-mm>.jsonl (schema run.schema.json, fields as in overall §3), always written, including on failure.

Stages 2, 5 and 8 can also run separately (fetch, parse --raw <path>, publish --records <path>), so a stage can be retried or replayed without the others (KN: independent Fetch, Parse, Load).

3. Adapter interface and registry

class Adapter(Protocol):
    id: str                  # "<family>"; one instance per AMC is configured from amcs.yaml
    doc_type: str            # "portfolio" | "factsheet" | "nav" | "scheme" | "aum" | "ter" | "performance" | "nfo" | "index"
    def fingerprint(self, raw: RawFile, cfg: dict) -> FingerprintResult: ...   # ok: bool, found: dict, expected: dict
    def parse(self, raw: RawFile, cfg: dict, ctx: ParseContext) -> list[dict]: ...  # envelope records, no I/O, no network

Adapters are pure: no network, no database, no clock (the context supplies as_of hints and the run id). That makes them testable from fixtures alone.

registry/amcs.yaml, one row per AMC:

Field	Example	Rule
amc_code	hdfc	lowercase slug, unique, stable
amfi_name	HDFC Mutual Fund	as in the AMFI scheme master
aum_rank	1	from AMFI AUM (S6); decides build order
portfolio.url / factsheet.url	URL or UNKNOWN	never guessed (data-source register rule)
portfolio.family / factsheet.family	cams_excel	a module in adapters/families/, or UNKNOWN
portfolio.config / factsheet.config	sheet pattern, header row, column map, units	family-specific; validated by the family
portfolio.fingerprint	{sheet_pattern, header_row, columns[], min_rows, max_rows}	stored from the last good file
status	planned / built / staging / production / blocked:<reason>	the measure for MC-2 to MC-4
terms	unchecked / allowed / manual-only	from the legal answer (for-atul 2026-10-25); manual-only disables scheduled fetch

4. Identity and mapping (TD-048)

• scheme_code: AMFI code, integer. isin: 12 characters, checked with the ISO 6166 check digit. amc_code: registry slug.
• A portfolio disclosure names a portfolio (usually the scheme name without plan/option). The adapter outputs portfolio_name exactly as disclosed, after trimming whitespace only. Nothing is normalised further at parse time.
• registry/portfolio_map.yaml rows: amc_code, portfolio_name, scheme_codes[], cardinality (1:1 | N:1), evidence (a disclosure file sha and a note), confirmed_by, confirmed_on.
• mapping.py proposes candidates (same AMC; scheme-master names that contain the disclosed name; same category). It never publishes on a proposal. Unmapped portfolios go to out/review/<run_id>.csv with their candidates. A person or the API agent confirms a row in the map by MR, and the next run publishes.
• portfolio_id in mm_data.mm_portfolio is the API's snapshot id. The framework never creates or assumes it (KN).

5. Common output schema v1 (TD-060)

Number rule: every NAV, money, percentage and ratio value is a decimal string ("123.4567"), never a float, so no precision is lost between the file, the adapter and the database. Dates are ISO YYYY-MM-DD; times are ISO with offset (+05:30). Money is in rupees: adapters convert lakh or crore using the family config's units.

Envelope (every record):

Field	Type	Req	Meaning
type	enum (record types below)	yes
schema	"mmcron/v1"	yes	bump on any breaking change
source_id	S1…S12, or AMC-<amc_code>	yes	data-source register
as_of	date	yes	business date of the observation (NAV date, month end of a disclosure)
source_ref	string	yes	locator in the raw file: sheet!row, page:line or line
data	object	yes	the type's fields below

Record types (data fields; ? = optional):

Type	Fields
nav	scheme_code int, nav dec>0, repurchase_price? dec, sale_price? dec
scheme	scheme_code int, scheme_name, amc_code, amfi_category, plan (direct/regular), option (growth/idcw/other), isin_growth?, isin_reinvest?, launch_date?, closure_date?, status (active/closed/matured)
aum	scope (scheme/amc), scheme_code?, amc_code, period (YYYY-MM or YYYY-Qn), aum_inr dec, average? bool
ter	scheme_code int, ter_regular_pct? dec, ter_direct_pct? dec, effective_from date
performance	scheme_code int, period (1y,3y,5y,10y,si), return_pct dec, benchmark_name?, benchmark_return_pct? dec
nfo	scheme_name, amc_code, open_date, close_date, category?, min_investment_inr? dec
index_value	symbol (exact, e.g. NIFTY 50), close dec>0, open?, high?, low? dec
holding	amc_code, portfolio_name, scheme_codes[] (filled by mapping), instrument_name, isin?, instrument_type (equity,debt,money_market,govt_security,reit_invit,derivative,cash_equivalent,other), industry? (as disclosed), rating?, quantity? dec, market_value_inr? dec, weight_pct dec
fund_details	amc_code, portfolio_name, scheme_codes[], managers[] (name, since? date), benchmark_name?, inception_date?, std_dev_pct?, sharpe?, beta?, portfolio_turnover? dec, riskometer? (low,low_to_moderate,moderate,moderately_high,high,very_high), exit_load_text?, fields_missing[]

fields_missing lists every optional field the source did not give. This lets "the factsheet doesn't say" be told apart from "the parser missed it" (TD-050). TER and AUM are not in fund_details: they come from AMFI (TD-054).

6. Validation rules (validate.py)

Rule	Applies to	Level
Schema valid (JSON Schema v1)	all	error
as_of not in the future; within the job's freshness window	all	error
No duplicate key in the record set: nav (scheme_code, as_of); index_value (symbol, as_of); holding (portfolio_name, isin or instrument_name); ter (scheme_code, effective_from)	as listed	error
nav, close > 0; day-on-day change ≤ 20% unless the scheme is flagged (merger, segregation)	nav, index_value	error above 20%, warning above 10%
ISIN check digit valid	holding, scheme	error
Sum of weight_pct per portfolio within 100 ± 0.5	holding	warning (derivatives and net current assets can push it out); error outside 100 ± 5
Row count within ±30% of the last good run for the same source	all	warning
scheme_code known in the scheme master	nav, ter, performance, mapped holding	error (unknown codes go to the review list)

7. Publish targets and CLI

Target	Writes	When
file (default)	out/publish/<run_id>.<type>.jsonl	S2, before ingest routes exist; tests
staging	POST {MMCRON_STAGING_API_URL}/api/ingest/<type>, header X-MM-Key: {MMCRON_STAGING_API_KEY}	from S3
production	the same with MMCRON_PRODUCTION_*	from S4; refuses unless MMCRON_ALLOW_PRODUCTION=1 is set in the host's environment (set only on the production host)

• Before any staging publish, the publisher runs the existing fail-closed preflight (staging_routing/preflight.py, KN), which becomes mmcron/preflight.py in the move.
• Batches of up to 1,000 records. A repeat of a run is safe because ingest is idempotent.
• No database driver is imported anywhere in mmcron/; a test enforces this.

CLI: python -m mmcron run <job_id> --env file|staging|production [--file <path>]; fetch <job_id>; parse <job_id> --raw <path>; publish --records <path> --env …; registry check; replay <raw path> (parse an archived file and compare with what was published).

8. Logging

Standard logging with a JSON-lines formatter to logs/mmcron.log (fields: time, level, job, run_id, stage, message). No secrets; keys are never logged, and a test checks a run's log against the key value. Run records are separate (§2.9) and are what the collector reads.

C4 element

• mmcronRegistry
• mmcronFetch
• mmcronRaw
• mmcronAdapters
• mmcronValidate
• mmcronPublish
• apiIngest

Evaluation

	Assessment
V value	High. It is the critical path for 51 AMCs by 28 Nov, ends silent failures, and gives every value a source. The same schema serves the API's ingest validation
S simplicity	One pipeline, pure adapters, YAML configuration, two dependencies. Families keep most AMCs to configuration rather than code
C cost	Sonnet: about 2–3 sessions for items 1–7. Harness: free or flash-lite per adapter. Disk for the raw archive: small (estimate 1–3 GB a year)
D data	Public data only (AMFI, AMC disclosures). Keys are in environment variables, never in files. No client data. Free models are allowed for adapters (ADR-0009)
L legal	Automated download depends on AMC terms (for-atul 2026-10-25). The terms field and --file let adapters be built and run on hand-saved files until the answer comes

Test plan

• Schema: for each record type, one valid example and at least three invalid ones (missing field, float instead of decimal string, bad enum) in tests/test_schema.py.
• Fingerprint stops the run: a fixture with a renamed column gives stopped-format, exit 3, an empty publish output and a run record with the diff (tests/test_pipeline.py).
• Validation stops the run: duplicate NAV keys and a bad ISIN give stopped-validation, exit 4, nothing published.
• Mapping: an unmapped portfolio goes to the review list and is not published; a confirmed map row publishes it.
• Pure adapters: adapter tests run with network and DB modules blocked (monkeypatched socket); importing a DB driver in mmcron/ fails a test.
• Idempotent and replayable: running the same fixture twice to the file target gives byte-identical output; replay on an archived file reproduces the published records.
• Acceptance (one test, every fixture): tests/test_fixtures.py runs every fixtures/<amc>/<type>/<month>/input.* through its family and compares with expected.json. A new adapter is done when its fixture passes.
• Registry: 51 AMC rows; unique amc_code; every non-UNKNOWN family exists; aum_rank unique.
• Secrets: a test sets a fake key, runs a publish against a local stub server, and asserts the key never appears in logs or run records.
• Windows: paths via pathlib; tests pass on the laptop runner (tags laptop, docker) and on Linux CI.
• First real adapters: three AMCs from different families (harness, items 9–11) prove the framework before the remaining 12 of the top 15 start.

Risk

risk: high

Why high: the schema is a contract between MM-Cron and the API's ingest routes, and it carries financial data to client-facing pages. A wrong unit (lakh against rupees) or a silent mapping error would publish wrong numbers. Mitigations: decimal strings, explicit units in family config, mapping never automatic, error-level validation, file target first, staging for a full cycle before production (overall §7).

The framework code itself has no production path until MC-4, so its build items are medium (integration branch, sprint batch approval). Enabling production publishing is a separate high-risk change (Lane P, Atul).

Work items

#	Title	Risk	Model	Scope paths	Depends on	Cx	Tests
1	Schema v1: JSON Schema files for envelope, run and the 9 record types; schema.validate()	medium	sonnet	owp/mm-crons/mmcron/schema/, owp/mm-crons/tests/test_schema.py		3	tests/test_schema.py
2	Adapter interface, RawFile, ParseContext, FingerprintResult	medium	sonnet	owp/mm-crons/mmcron/adapters/base.py, owp/mm-crons/tests/test_adapter_base.py	1	2	tests/test_adapter_base.py
3	Fetcher and raw archive with manifest	medium	sonnet	owp/mm-crons/mmcron/fetch.py, owp/mm-crons/mmcron/archive.py, owp/mm-crons/tests/test_fetch_archive.py		2	tests/test_fetch_archive.py
4	Validator: the §6 rules, error and warning levels	medium	sonnet	owp/mm-crons/mmcron/validate.py, owp/mm-crons/tests/test_validate.py	1	3	tests/test_validate.py
5	Mapping and review list; portfolio_map.yaml loader	medium	sonnet	owp/mm-crons/mmcron/mapping.py, owp/mm-crons/registry/portfolio_map.yaml, owp/mm-crons/tests/test_mapping.py	1	2	tests/test_mapping.py
6	Publisher (file, staging, production guard), preflight hook, no-DB-driver guard	medium	sonnet	owp/mm-crons/mmcron/publish.py, owp/mm-crons/tests/test_publish.py	1	3	tests/test_publish.py
7	Pipeline, run record, CLI, JSON logging, fixture acceptance test	medium	sonnet	owp/mm-crons/mmcron/run.py, owp/mm-crons/mmcron/main.py, owp/mm-crons/tests/test_pipeline.py, owp/mm-crons/tests/test_fixtures.py	2,3,4,5,6	3	tests/test_pipeline.py, tests/test_fixtures.py
8	Registry: 51 AMC rows (amfi_name, aum_rank, URLs from #64 or UNKNOWN, format family), registry check	low	chatgpt-api-mmcron	owp/mm-crons/registry/amcs.yaml, owp/mm-crons/mmcron/registry.py	7	2	tests/test_registry.py
9	First family adapter: portfolio disclosure, AMC rank 1 (Excel), fixture and expected output	low	gemini-flash-lite	owp/mm-crons/mmcron/adapters/families/, owp/mm-crons/tests/fixtures/	7,8	2	tests/test_fixtures.py
10	Second family adapter: portfolio disclosure, first AMC of a different family	low	gemini-flash-lite	owp/mm-crons/mmcron/adapters/families/, owp/mm-crons/tests/fixtures/	9	2	tests/test_fixtures.py
11	AMFI NAV daily adapter (amfi_text family)	low	gemini-flash-lite	owp/mm-crons/mmcron/adapters/families/amfi_text.py, owp/mm-crons/tests/fixtures/amfi/	7	2	tests/test_fixtures.py
12	Move owp/mm-crons/ into moneymarche/mm-crons (one MR; staging_routing/preflight.py becomes mmcron/preflight.py; CI job on the laptop runner)	medium	chatgpt-api-mmcron	mm-crons: mmcron/, registry/, tests/, .gitlab-ci.yml	7	2	mm-crons CI test
13	Adapter issues for AMC ranks 2–15 (one per AMC, portfolio), each with sample file and expected output	low	chatgpt-api-mmcron	mm-crons issues	10	1	per issue