Feature tickerplant

di/tickerplant/VERSION

@@ -0,0 +1 @@
+0.1.0

di/tickerplant/deps.q

@@ -0,0 +1,4 @@
+/ hard module dependencies and their minimum versions, validated by di.depcheck.
+/ these are the modules di.tickerplant `use`s in init.q. the injected log and timer are not declared
+/ here - di.depcheck validates them through its core-contract check.
+deps:`di.pubsub`di.eodtime`di.tplog!("0.1.0";"0.1.0";"0.1.0");

di/tickerplant/init.q

@@ -0,0 +1,11 @@
+/ di.tickerplant - tick-capture: log and publish incoming updates, roll at end of day
+/ hard dependencies - imported here as module-local handles (before the impl loads), used by tickerplant.q
+pubsub:use`di.pubsub
+eodtime:use`di.eodtime
+tplog:use`di.tplog
+\l ::tickerplant.q
+/ module version, read from the VERSION file (one plain-text file to bump per release). read
+/ module-relative at load (`:::` resolves to di/tickerplant) and BEFORE export, since export:([...])
+/ evaluates each name; version stays in the export so di.depcheck reads it from the export dict
+version:trim first read0`:::VERSION
+export:([init;upd;subscribe;endofday;getcounts;gettables;getapimeta;version])

di/tickerplant/test.csv

@@ -0,0 +1,35 @@
+action,ms,bytes,lang,code,repeat,minver,comment
+before,0,0,q,tp:use`di.tickerplant,1,,load the module under test
+before,0,0,q,timer:use`di.timer,1,,real di.timer (addjob works without timer.init - no live .z.ts)
+before,0,0,q,ntp:use`di.tplog,1,,di.tplog, to replay a tickerplant-written log
+before,0,0,q,eod:use`di.eodtime,1,,di.eodtime, to force a roll due in a test
+before,0,0,q,os:use`di.os,1,,os module for portable path resolution
+before,0,0,q,"system ""l "",os.abspath[""di/tickerplant/test.q""]",1,,load fixture helpers
+before,0,0,q,setupfixture[],1,,create the temp fixture root
+comment,,,,,,,module metadata - version and getapimeta
+true,0,0,q,10h=type tp`version,1,1,version is a string
+true,0,0,q,0<count tp`version,1,1,version is non-empty
+true,0,0,q,`name`public`descrip`params`return~cols tp[`getapimeta][],1,1,getapimeta rows carry the registry columns
+true,0,0,q,(asc (key tp) except `init`getapimeta`version)~asc exec name from tp[`getapimeta][],1,1,getapimeta documents exactly the callable exports
+true,0,0,q,not any `init`getapimeta`version in exec name from tp[`getapimeta][],1,1,plumbing and version are not registered in the api
+comment,,,,,,,init dependency validation
+fail,0,0,q,tp.init[(::)],1,1,init rejects a non-dict deps
+fail,0,0,q,tp.init depsnolog[],1,1,init rejects deps without a log key
+fail,0,0,q,tp.init depsonlylog[],1,1,init rejects deps without a timer
+fail,0,0,q,tp.init depsbadtimer[],1,1,init rejects a timer that does not expose addjob
+fail,0,0,q,tp.init depsnoschemas[],1,1,init rejects deps without schemas
+comment,,,,,,,init / lifecycle
+true,0,0,q,testinit[],1,1,init materialises root tables (g# sym) and schedules the timer job
+true,0,0,q,testupdbatch[],1,1,batch upd stamps, buffers and logs
+true,0,0,q,testemptyupd[],1,1,an empty update is a no-op (no throw)
+true,0,0,q,testemptyupdtriggersroll[],1,1,an empty update still triggers an overdue roll
+true,0,0,q,testzerolatency[],1,1,zero-latency upd publishes without buffering
+true,0,0,q,testendofday[],1,1,endofday flushes, rolls the log and resets counts
+true,0,0,q,testreinitnoleak[],1,1,re-init closes the old log handle (no fd leak)
+comment,,,,,,,di.tplog integration
+true,0,0,q,testlogroundtrip[],1,1,a tickerplant-written log replays through di.tplog
+true,0,0,q,testcheckrepaironroll[],1,1,rolling into a corrupt log repairs it via di.tplog.check
+comment,,,,,,,upd input validation
+fail,0,0,q,tp[`upd][1;()],1,1,upd rejects a non-symbol table
+fail,0,0,q,tp[`upd][`nope;()],1,1,upd rejects an unknown table
+after,0,0,q,teardownfixture[],1,,remove the temp fixture root

di/tickerplant/test.q

@@ -0,0 +1,131 @@
+/ fixture helpers for di.tickerplant's k4unit tests.
+/ tickerplant orchestrates the real di.pubsub / di.eodtime / di.tplog / di.timer modules with an
+/ injected log, so these tests wire the real modules (tplog resolved via QPATH) and drive a full
+/ init -> upd -> roll cycle. the data tables live at ROOT (the tickerplant owns them and di.pubsub
+/ reads them by name), and replay runs the ROOT upd, so a schema + recorder upd live here at root.
+/ tp (di.tickerplant), timer (di.timer) and ntp (di.tplog) are bound by test.csv's before rows.
+
+base:"/tmp/di_tickerplant_k4unit";
+trade:([]time:`timestamp$();sym:`symbol$();price:`float$();size:`long$());
+upd:{[t;x] t insert x};
+
+/ capturing logger shared by tickerplant and its dep modules, so any module's output is assertable
+logcap:([]lvl:`symbol$();ctx:`symbol$();msg:());
+caplog:{`info`warn`error!(
+  {[c;m]`logcap insert(`info;c;m);};
+  {[c;m]`logcap insert(`warn;c;m);};
+  {[c;m]`logcap insert(`error;c;m);})};
+
+freshdir:{[sub] dd:base,"/",sub; system"rm -rf ",dd; system"mkdir -p ",dd; dd};
+
+/ deps for init - real timer, capturing log, one trade table, logging into dd, batch flag as given
+mkdeps:{[dd;batch] `log`timer`schemas`logdir`logname`batch!(caplog[];timer;enlist[`trade]!enlist trade;dd;"tp";batch)};
+
+/ fresh init into a clean per-test dir; clears the root table and the log capture first
+freshinit:{[sub;batch]
+  dd:freshdir sub;
+  `trade set 0#trade;
+  `logcap set 0#logcap;
+  tp.init mkdeps[dd;batch];
+  dd};
+
+/ a feed update: sym, price, size - no time, so the tickerplant stamps it
+row:{[s] (s;1.0;100)};
+
+setupfixture:{system"rm -rf ",base; system"mkdir -p ",base;};
+teardownfixture:{system"rm -rf ",base;};
+
+/ deps builders for the init validation fail rows
+depsnolog:{(enlist`x)!enlist 1};
+depsonlylog:{enlist[`log]!enlist caplog[]};
+depsnoschemas:{`log`timer!(caplog[];timer)};
+depsbadtimer:{`log`timer`schemas!(caplog[];(enlist`x)!enlist 1;enlist[`trade]!enlist trade)};
+
+/ =============================================================================
+/ tests (each returns 1b on success)
+/ =============================================================================
+
+/ init materialises the tables at root (g# on sym), schedules the timer job, zeroes the counts
+testinit:{[]
+  freshinit["init";1b];
+  c:tp[`getcounts][];
+  (`g=attr exec sym from trade) and (enlist[`trade]~tp[`gettables][]) and (0=c`i) and (0=c`j)
+    and (-14h=type c`d) and `tickerplant in exec id from timer.getalljobs[]};
+
+/ batch mode: upd stamps, buffers into the root table, and logs (bumping j; i unchanged)
+testupdbatch:{[]
+  freshinit["updb";1b];
+  tp[`upd][`trade;row`AAPL];
+  tp[`upd][`trade;row`MSFT];
+  c:tp[`getcounts][];
+  (2=count trade) and (`AAPL`MSFT~exec sym from trade) and (not any null exec time from trade)
+    and (2=c`j) and 0=c`i};
+
+/ an empty update is a no-op: no throw, nothing buffered, nothing logged
+testemptyupd:{[]
+  freshinit["empty";1b];
+  tp[`upd][`trade;()];
+  c:tp[`getcounts][];
+  (0=count trade) and 0=c`j};
+
+/ an empty update still triggers an overdue roll (the roll check runs before the empty-data skip)
+testemptyupdtriggersroll:{[]
+  freshinit["emptyroll";1b];
+  oldd:first tp[`getcounts][]`d;
+  eod.setnextroll .z.p-0D01:00:00;
+  tp[`upd][`trade;()];
+  (oldd+1)=first tp[`getcounts][]`d};
+
+/ zero-latency mode: upd publishes immediately, does NOT buffer into the root table, still logs
+testzerolatency:{[]
+  freshinit["zl";0b];
+  tp[`upd][`trade;row`AAPL];
+  c:tp[`getcounts][];
+  (0=count trade) and 1=c`j};
+
+/ re-init is safe: it closes the previous log handle instead of leaking the descriptor, and logging
+/ keeps working. fd count (linux /proc, as the suite is already unix-coupled) must not grow.
+testreinitnoleak:{[]
+  fddir:"/proc/",(string .z.i),"/fd";
+  freshinit["reinit";1b];
+  b:"J"$first system"ls ",fddir," | wc -l";   / baseline AFTER the first init - one log handle is open
+  freshinit["reinit";1b]; freshinit["reinit";1b]; freshinit["reinit";1b];   / re-init must not add fds
+  a:"J"$first system"ls ",fddir," | wc -l";
+  tp[`upd][`trade;row`AAPL];
+  (a=b) and 1=tp[`getcounts][]`j};
+
+/ the tp log tickerplant writes replays through di.tplog - the two modules agree on the log format
+testlogroundtrip:{[]
+  dd:freshinit["rt";1b];
+  tp[`upd][`trade;row`AAPL];
+  tp[`upd][`trade;row`MSFT];
+  lf:hsym`$dd,"/tp",string first tp[`getcounts][]`d;
+  `trade set 0#trade; `rcv set 0;
+  `upd set {[t;x] `rcv set rcv+1; t insert x;};
+  n:ntp[`replay] lf;
+  `upd set {[t;x] t insert x;};
+  (2=n) and (2=rcv) and 2=count trade};
+
+/ endofday flushes the buffer, rolls to the next day's log, and resets the counts
+testendofday:{[]
+  dd:freshinit["eod";1b];
+  tp[`upd][`trade;row`AAPL];
+  oldd:first tp[`getcounts][]`d;
+  tp[`endofday][];
+  c:tp[`getcounts][];
+  ((oldd+1)=c`d) and (0=c`i) and (0=c`j) and (0=count trade)
+    and not ()~key hsym`$dd,"/tp",string oldd+1};
+
+/ rolling into a pre-existing CORRUPT log makes openlog repair it via di.tplog.check
+testcheckrepaironroll:{[]
+  dd:freshinit["rep";1b];
+  oldd:first tp[`getcounts][]`d;
+  nl:hsym`$dd,"/tp",string oldd+1;
+  h:hopen nl;
+  h enlist (`upd;`trade;(enlist 2026.08.13D10:00;enlist`AAPL;enlist 1.0;enlist 100));
+  h enlist (`upd;`trade;(enlist 2026.08.13D10:01;enlist`IBM;enlist 2.0;enlist 200));
+  hclose h;
+  nl set (-8)_read1 nl;
+  `logcap set 0#logcap;
+  tp[`endofday][];
+  (not ()~key hsym`$dd,"/tp",(string oldd+1),".good") and `warn in exec lvl from logcap where ctx=`check};

di/tickerplant/tickerplant.md

@@ -0,0 +1,96 @@
+# di.tickerplant
+
+Core tick-capture for the modular TorQ world: receive updates from feeds, stamp them, write them to a
+tickerplant log for recovery, and publish them to subscribers, rolling the log at end of day. It is
+the modular replacement for TorQ's `code/processes/tickerplant.q`.
+
+It orchestrates three hard dependencies — `di.pubsub` (subscribe/publish), `di.eodtime` (roll timing)
+and `di.tplog` (log check/repair) — with an injected logger and timer.
+
+## Import and init
+
+```q
+tp:use`di.tickerplant
+
+trade:([]time:`timestamp$();sym:`symbol$();price:`float$();size:`long$())
+tp.init[`log`timer`schemas!(logdep;timerdep;enlist[`trade]!enlist trade)]
+```
+
+`init` takes a single deps dict:
+
+| Key | Required | Description |
+|---|---|---|
+| `log` | yes | `` `info`warn`error `` dict of `{[ctx;msg]}` functions |
+| `timer` | yes | `di.timer`'s exports (must expose `addjob`) |
+| `schemas` | yes | `tablename!schema` dict; the tables to capture |
+| `batch` | no | `1b` (default) buffers and publishes on a timer; `0b` publishes each update immediately |
+| `batchperiod` | no | batch publish interval (timespan, whole seconds; default `0D00:00:01`) |
+| `logdir` | no | directory for the tp log; `""` (default) disables logging |
+| `logname` | no | log filename prefix (default `"tp"`; file is `<logdir>/<logname><date>`) |
+| `subtables` | no | tables offered for subscription (default: all captured tables) |
+| `rolltimezone` / `datatimezone` / `rolltimeoffset` | no | forwarded to `di.eodtime` |
+
+`init` initialises the dependency modules (`di.eodtime`, `di.tplog`, `di.pubsub`), materialises the
+schemas as root tables (applying `` `g# `` to any `sym` column), opens today's log, and schedules a
+single timer job that flushes the buffer (batch mode) and checks for the end-of-day roll. It is
+idempotent — a re-init does not re-add the timer job.
+
+## Root tables and the upd contract
+
+The captured tables live at **root**, not in `.z.m`: a tickerplant owns its tables, feeds insert into
+them, and `di.pubsub` reads them by name, so they cannot be module-local. This is the one deliberate
+root-state exception; all other mutable state is module-local. `di.torq` wires the process's root
+`upd` to `tickerplant.upd` so feeds can publish to it.
+
+## Exported functions
+
+| Function | Signature | Description |
+|---|---|---|
+| `upd` | `[table;data]` | Feed entry point: stamp the update, then buffer+log (batch) or publish+log (zero-latency). |
+| `subscribe` | `[tables;filters]` | Register a subscriber (delegates to `di.pubsub`); called by downstream processes over IPC. |
+| `endofday` | `[]` | Flush the buffer, notify subscribers, roll the tp log, and advance the end-of-day state. |
+| `getcounts` | `[]` | `` `i`j`d `` — messages published (`i`), messages logged (`j`), and the trading date. |
+| `gettables` | `[]` | The tables this tickerplant captures. |
+
+`getapimeta[]` and `version` are also exported, as metadata for `di.torq` / `di.depcheck`.
+
+## Dependencies
+
+Hard (imported via `use` in `init.q`, declared in `deps.q`): `di.pubsub`, `di.eodtime`, `di.tplog`.
+Injected via `init`: `log` and `timer` (both required; validated by `di.depcheck`'s contract check,
+not declared in `deps.q`).
+
+`di.tplog` is used only for check/repair on recovery — when `openlog` finds a pre-existing log it runs
+it through `tplog.check`, repairing a corrupt one. The tickerplant appends to and rolls the log itself
+(opening for append, not replaying), since `di.tplog`'s `open`/`roll` replay through `upd`, which a
+tickerplant must not do to its own log.
+
+## Design notes
+
+- **Batch vs zero-latency.** In batch mode `upd` inserts into the root table and the timer job
+  publishes the accumulated rows every `batchperiod`, then clears them. In zero-latency mode `upd`
+  publishes each update immediately and does not buffer. Both modes log every message.
+- **End of day.** The roll fires when the current time passes `di.eodtime`'s next roll timestamp,
+  checked on every `upd` and on every timer tick. `endofday` flushes, notifies subscribers, rolls the
+  log to the next day, and refreshes the roll time and data-timestamp offset from `di.eodtime`.
+- **No `di.handlers` dependency.** Subscriber-disconnect cleanup is handled by `di.pubsub`'s own
+  `.z.pc`. (`di.pubsub` should migrate to `di.handlers` so `.z.*` is not assigned outside the central
+  registry — tracked separately, out of scope here.)
+
+## Testing
+
+`test.csv` / `test.q` (k4unit) run against the **real** `di.pubsub`, `di.eodtime`, `di.tplog` and
+`di.timer` — no dependencies are mocked (the timer is used without `init`, so no live `.z.ts`, and its
+job is exercised through `endofday`). A capturing logger is shared across the modules so their output
+is assertable.
+
+Coverage: the metadata/version contract; strict `init` dependency validation (a `fail` row per guard);
+init materialising the root tables and scheduling the timer job; batch and zero-latency `upd`;
+`endofday` flushing and rolling; `upd` input validation; and the two `di.tplog` integration points — a
+tickerplant-written log replaying through `di.tplog`, and rolling into a corrupt log repairing it via
+`tplog.check`.
+
+```q
+k4unit:use`di.k4unit
+k4unit.moduletest`di.tickerplant
+```