Default-off external cash-flow facts and Modified Dietz returns

tests/console_performance_stats_validation.mjs

@@ -1,5 +1,7 @@
 import "./console_equity_history_validation.mjs";
 import assert from "node:assert/strict";
+import "./console_external_flow_validation.mjs";
+import "./external_flow_receiver_validation.mjs";
 import {
   businessDate,
   calendarCells,

tests/external_flow_receiver_validation.mjs

@@ -0,0 +1,84 @@
+import assert from "node:assert/strict";
+import { createRequire } from "node:module";
+import { fileURLToPath } from "node:url";
+import worker, { __test } from "../web/strategy-switch-console/worker.js";
+
+const require = createRequire(new URL("../web/strategy-switch-console/package.json", import.meta.url));
+const { Miniflare } = require(process.env.QRT_MINIFLARE_MODULE || "miniflare");
+const hash = "a".repeat(64);
+const now = Date.now() - 120000;
+const at = offset => new Date(now + offset).toISOString();
+const start = at(-86400000), end = at(0);
+const base = {
+  schema_version: "longbridge_account_snapshot_history.v1", snapshot_schema_version: "longbridge_account_snapshot.v1",
+  account_scope: "PAPER", target_id: "synthetic-target", snapshot_atomic: false,
+  source_binding: { kind: "deployment_scope_token_version", status: "bound", id: hash },
+  observed_started_at: start, observed_finished_at: start, observation_date: start.slice(0, 10),
+  broker_reported_balances: [{ currency: "USD", net_assets: "100", total_cash: "100" }],
+  cash: [{ currency: "USD", available_cash: "100", frozen_cash: "0", settling_cash: "0" }],
+};
+const flow = { schema_version: "external_flow.v1", windows: [{ currency: "USD", date: end.slice(0, 10), period_start: start, period_end: end,
+  source: "longbridge_cashflow", coverage: "complete", records: [{ source_ref: "b".repeat(64), type: "deposit", amount: "100",
+    effective_at: at(-43200000), settlement_date: null, transfer_ref: null }] }] };
+const latest = { ...base, observed_started_at: end, observed_finished_at: end, observation_date: end.slice(0, 10),
+  broker_reported_balances: [{ currency: "USD", net_assets: "215", total_cash: "215" }],
+  cash: [{ currency: "USD", available_cash: "215", frozen_cash: "0", settling_cash: "0" }], externalFlow: flow };
+const settings = { ACCOUNT_FACTS_READ_MODEL_ENABLED: "true", CONSOLE_EXTERNAL_FLOW_ENABLED: "true",
+  ACCOUNT_FACTS_SYNC_TOKEN: "synthetic-sync-token", SESSION_SECRET: "synthetic-session", ALLOWED_GITHUB_LOGINS: "synthetic-reader" };
+let outbound = 0;
+const mf = new Miniflare({ modules: true, modulesRules: [{ type: "ESModule", include: ["**/*.js"] }],
+  scriptPath: fileURLToPath(new URL("../web/strategy-switch-console/worker.js", import.meta.url)),
+  compatibilityDate: "2026-06-08", bindings: settings,
+  durableObjects: { STRATEGY_SWITCH_RUNTIME_INSTANCES: { className: "RuntimeInstances", useSQLite: true } },
+  kvNamespaces: ["STRATEGY_SWITCH_CONFIG"],
+  outboundService: () => { outbound++; throw new Error("external calls forbidden"); },
+});
+try {
+  const kv = await mf.getKVNamespace("STRATEGY_SWITCH_CONFIG");
+  const namespace = await mf.getDurableObjectNamespace("STRATEGY_SWITCH_RUNTIME_INSTANCES");
+  const option = { key: "synthetic", label: "Synthetic", target_name: "synthetic-target", service_name: "synthetic-service",
+    deployment_selector: "synthetic-deployment", account_scope: "paper", account_selector: "PAPER", supported_domains: ["us_equity"] };
+  const binding = { ...option, platform: "longbridge", account_key: option.key, target_id: "synthetic-target",
+    source_binding: { kind: "deployment_scope_token_version", id: hash } };
+  delete binding.key; delete binding.label; delete binding.supported_domains;
+  await kv.put("account_options", JSON.stringify({ longbridge: [option] }));
+  await kv.put("account_facts_bindings", JSON.stringify({ schema_version: "qsl_account_facts_bindings.v1", bindings: [binding] }));
+  const env = { ...settings, STRATEGY_SWITCH_CONFIG: kv, STRATEGY_SWITCH_RUNTIME_INSTANCES: namespace };
+  const cookie = await __test.makeSession("synthetic-reader", [], settings);
+  const post = (body, token = settings.ACCOUNT_FACTS_SYNC_TOKEN, override = {}) => worker.fetch(new Request("https://synthetic.example/api/account-facts/sync", {
+    method: "POST", headers: { Authorization: `Bearer ${token}`, "Content-Type": "application/json" }, body: JSON.stringify(body),
+  }), { ...env, ...override });
+  const accept = async body => { const r = await post(body); assert.equal(r.status, 200, await r.clone().text()); return r.json(); };
+  const get = async (path, override = {}) => {
+    const r = await worker.fetch(new Request(`https://synthetic.example${path}`, { headers: { Cookie: `qsl_switch_session=${cookie}` } }), { ...env, ...override });
+    assert.equal(r.status, 200, await r.clone().text()); return r.json();
+  };
+  const historyPath = "/api/account-facts/history?platform=longbridge&account_key=synthetic&currency=USD";
+  const summaryPath = "/api/performance/summary?scope=synthetic&platform=longbridge";
+  await accept(base);
+  const { externalFlow, ...legacyLatest } = latest;
+  await accept(legacyLatest);
+  const before = await get(summaryPath);
+  const offBefore = await get(summaryPath, { CONSOLE_EXTERNAL_FLOW_ENABLED: "false" });
+  assert.deepEqual(before, offBefore, "ON no source equals OFF through HTTP adapter");
+  assert.equal((await post(latest, "wrong-synthetic-token")).status, 401);
+  assert.equal((await post(latest, settings.ACCOUNT_FACTS_SYNC_TOKEN, { CONSOLE_EXTERNAL_FLOW_ENABLED: "false" })).status, 409);
+  await accept(latest);
+  assert.equal((await accept(latest)).unchanged, true);
+  assert.deepEqual((await get(historyPath)).externalFlow, flow);
+  const summary = await get(summaryPath);
+  assert.ok(Math.abs(summary.twr_return_percent - 10) < 1e-10, JSON.stringify(summary));
+  assert.equal(summary.cashflow_per_currency[0].approximate, true);
+  assert.deepEqual(await get(summaryPath, { CONSOLE_EXTERNAL_FLOW_ENABLED: "false" }), offBefore, "OFF read ignores stored extension");
+  const conflict = structuredClone(latest); conflict.externalFlow.windows[0].records[0].amount = "101";
+  assert.equal((await post(conflict)).status, 409);
+  assert.deepEqual((await get(historyPath)).externalFlow, flow, "conflict cannot change SQLite history");
+  const malformed = structuredClone(latest); malformed.externalFlow.windows[0].records[0].source_ref = "raw-private-reference";
+  assert.equal((await post(malformed)).status, 400);
+  const wrong = structuredClone(latest); wrong.source_binding.id = "c".repeat(64);
+  assert.equal((await post(wrong)).status, 409);
+  const oversize = structuredClone(latest); oversize.externalFlow.windows[0].records = Array(257).fill(flow.windows[0].records[0]);
+  assert.equal((await post(oversize)).status, 400);
+  assert.equal(outbound, 0);
+  console.log("external_flow_receiver_validation: ok (authenticated synthetic HTTP -> SQLite -> history -> Dietz; no outbound calls)");
+} finally { await mf.dispose(); }

web/strategy-switch-console/application/account_facts/external_flow.js

@@ -0,0 +1,110 @@
+// Private account-facts extension. Never infer complete coverage from an empty list.
+import { dateMs } from "../../performance_math.js";
+
+export const EXTERNAL_FLOW_SCHEMA = "external_flow.v1";
+const DAY = 86400000;
+const SOURCES = new Set(["ibkr_flex", "schwab_transactions", "longbridge_cashflow", "binance_history"]);
+const TYPES = new Set(["deposit", "withdrawal", "transfer_in", "transfer_out"]);
+const HASH = /^[a-f0-9]{64}$/;
+const MONEY = /^-?(?:0|[1-9]\d{0,11})(?:\.\d{1,8})?$/;
+export const externalFlowEnabled = (env) => env?.CONSOLE_EXTERNAL_FLOW_ENABLED === "true";
+export function flowError(code = "invalid_external_flow", status = 400) {
+  const error = new Error(code);
+  error.code = code;
+  error.status = status;
+  return error;
+}
+function exact(value, keys) {
+  return value && typeof value === "object" && !Array.isArray(value)
+    && Object.keys(value).length === keys.length && keys.every(k => Object.hasOwn(value, k));
+}
+export function instant(value) {
+  if (typeof value !== "string" || !/^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d{1,3})?Z$/.test(value)) return NaN;
+  const ms = Date.parse(value);
+  return Number.isFinite(ms) && new Date(ms).toISOString().slice(0, 19) === value.slice(0, 19) ? ms : NaN;
+}
+const windowKey = w => `${w.currency}:${w.period_start}:${w.period_end}`;
+const stable = value => JSON.stringify(value);
+
+export function normalizeExternalFlow(raw, { now = Date.now(), maxWindows = 31, maxRecords = 256 } = {}) {
+  if (!exact(raw, ["schema_version", "windows"]) || raw.schema_version !== EXTERNAL_FLOW_SCHEMA
+      || !Array.isArray(raw.windows) || !raw.windows.length || raw.windows.length > maxWindows
+      || new TextEncoder().encode(stable(raw)).length > (maxWindows === 31 ? 32768 : 524288)) throw flowError();
+  let count = 0;
+  const seen = new Map();
+  const keys = new Set();
+  const windows = raw.windows.map(w => {
+    if (!exact(w, ["currency", "date", "period_start", "period_end", "coverage", "source", "records"])
+        || !/^[A-Z][A-Z0-9]{2,9}$/.test(w.currency) || !Number.isFinite(dateMs(w.date))
+        || !SOURCES.has(w.source) || !["complete", "incomplete"].includes(w.coverage)
+        || !Array.isArray(w.records)) throw flowError();
+    const start = instant(w.period_start), end = instant(w.period_end);
+    if (!Number.isFinite(start) || !Number.isFinite(end) || end <= start || end - start > 2 * DAY
+        || end > now || w.date !== w.period_end.slice(0, 10) || keys.has(windowKey(w))) throw flowError();
+    keys.add(windowKey(w));
+    const refs = new Map();
+    for (const r of w.records) {
+      if (++count > maxRecords || !exact(r, ["source_ref", "type", "amount", "effective_at", "settlement_date", "transfer_ref"])
+          || !HASH.test(r.source_ref) || !TYPES.has(r.type) || typeof r.amount !== "string" || !MONEY.test(r.amount)
+          || !Number.isFinite(Number(r.amount)) || Number(r.amount) === 0
+          || ((r.type === "deposit" || r.type === "transfer_in") !== (Number(r.amount) > 0))
+          || (r.settlement_date !== null && !Number.isFinite(dateMs(r.settlement_date)))
+          || (r.transfer_ref !== null && !HASH.test(r.transfer_ref))) throw flowError();
+      const at = instant(r.effective_at);
+      if (!Number.isFinite(at) || at <= start || at > end) throw flowError();
+      const record = { source_ref: r.source_ref, type: r.type, amount: r.amount,
+        effective_at: r.effective_at, settlement_date: r.settlement_date, transfer_ref: r.transfer_ref };
+      const prior = seen.get(r.source_ref);
+      const identity = stable({ currency: w.currency, source: w.source, ...record });
+      if (prior && prior !== identity) throw flowError("external_flow_conflict", 409);
+      seen.set(r.source_ref, identity);
+      refs.set(r.source_ref, record);
+    }
+    return { currency: w.currency, date: w.date, period_start: w.period_start, period_end: w.period_end,
+      coverage: w.coverage, source: w.source,
+      records: [...refs.values()].sort((a, b) => a.source_ref.localeCompare(b.source_ref)) };
+  }).sort((a, b) => windowKey(a).localeCompare(windowKey(b)));
+  for (let i = 1; i < windows.length; i++) {
+    const a = windows[i], b = windows[i - 1];
+    if (a.currency === b.currency && instant(a.period_start) < instant(b.period_end)
+        && instant(b.period_start) < instant(a.period_end)) throw flowError("external_flow_overlap", 409);
+  }
+  return { schema_version: EXTERNAL_FLOW_SCHEMA, windows };
+}
+
+export function mergeExternalFlow(prior, incoming) {
+  const rows = new Map((prior?.windows || []).map(w => [windowKey(w), w]));
+  for (const w of incoming.windows) {
+    const old = rows.get(windowKey(w));
+    if (old && stable(old) !== stable(w)) {
+      // Coverage may be completed later, but facts already accepted are immutable.
+      if (old.coverage !== "incomplete" || w.source !== old.source
+          || old.records.some(r => !w.records.some(n => stable(n) === stable(r)))) throw flowError("external_flow_conflict", 409);
+    }
+    rows.set(windowKey(w), w);
+  }
+  const newest = Math.max(...[...rows.values()].map(w => dateMs(w.date)));
+  const windows = [...rows.values()].filter(w => dateMs(w.date) >= newest - 365 * DAY);
+  return normalizeExternalFlow({ schema_version: EXTERNAL_FLOW_SCHEMA, windows }, { maxWindows: 4096, maxRecords: 4096 });
+}
+
+export function initializeExternalFlow(sql) {
+  sql.exec("CREATE TABLE IF NOT EXISTS account_external_flow (platform TEXT NOT NULL, account_key TEXT NOT NULL, context_json TEXT NOT NULL, payload_json TEXT NOT NULL, PRIMARY KEY(platform, account_key))");
+}
+const identity = c => stable([c.target_id, c.source_binding_id, c.account_scope, c.account_selector,
+  c.broker_account_hash ?? null, c.broker_account_id ?? null]);
+export function readExternalFlow(sql, context) {
+  const row = sql.exec("SELECT context_json, payload_json FROM account_external_flow WHERE platform = ? AND account_key = ?",
+    context.platform, context.account_key).toArray()[0];
+  if (!row || row.context_json !== identity(context)) return null;
+  return normalizeExternalFlow(JSON.parse(row.payload_json), { maxWindows: 4096, maxRecords: 4096 });
+}
+export function putExternalFlow(sql, context, raw) {
+  const incoming = normalizeExternalFlow(raw);
+  const prior = readExternalFlow(sql, context);
+  const merged = mergeExternalFlow(prior, incoming);
+  if (stable(prior) === stable(merged)) return { unchanged: true };
+  sql.exec("INSERT INTO account_external_flow (platform, account_key, context_json, payload_json) VALUES (?, ?, ?, ?) ON CONFLICT(platform, account_key) DO UPDATE SET context_json = excluded.context_json, payload_json = excluded.payload_json",
+    context.platform, context.account_key, identity(context), stable(merged));
+  return { unchanged: false };
+}

web/strategy-switch-console/application/account_facts/read.js

@@ -7,6 +7,7 @@
  * does not import worker.js (avoids cycles) and wire stays identical.
  */
 
+import { externalFlowEnabled } from "./external_flow.js";
 import {
   ACCOUNT_FACTS_HISTORY_MAX_DAYS,
   ACCOUNT_FACTS_SUPPORTED_PLATFORMS,
@@ -156,6 +157,7 @@ export async function accountFactsHistoryResponse(request, env, url, deps) {
     (item) => item.platform === platform && item.account_key === accountKey,
   ) || null;
   let days = [];
+  let externalFlow = null;
   if (option && binding && accountFactsOptionMatchesBinding(option, binding)) {
     try {
       const listed = await runtimeInstanceCommand(env, {
@@ -174,6 +176,7 @@ export async function accountFactsHistoryResponse(request, env, url, deps) {
         source_binding_id: binding.source_binding.id,
       });
       days = Array.isArray(listed?.days) ? listed.days : [];
+      if (externalFlowEnabled(env)) externalFlow = listed?.externalFlow || null;
     } catch (error) {
       return accountFactsErrorResponse(error);
     }
@@ -191,5 +194,6 @@ export async function accountFactsHistoryResponse(request, env, url, deps) {
     ...model,
     enabled: true,
     retention_days: ACCOUNT_FACTS_HISTORY_MAX_DAYS,
+    ...(externalFlowEnabled(env) && externalFlow ? { externalFlow } : {}),
   });
 }

web/strategy-switch-console/application/account_facts/sync.js

@@ -7,6 +7,7 @@
  * the DO account_facts_put command (unchanged).
  */
 
+import { externalFlowEnabled, normalizeExternalFlow } from "./external_flow.js";
 import {
   ACCOUNT_FACTS_MAX_BODY_BYTES,
   ACCOUNT_FACTS_RETURN_UNAVAILABLE,
@@ -90,6 +91,19 @@ export async function syncAccountFactsResponse(request, env, deps) {
   if (payloadPlatform !== authorizedPlatform) {
     return json({ ok: false, error: "account_facts_sync_token_platform_mismatch" }, 403);
   }
+  let externalFlow;
+  if (raw && Object.hasOwn(raw, "externalFlow")) {
+    if (!externalFlowEnabled(env)) return json({ ok: false, error: "external_flow_disabled" }, 409);
+    try {
+      externalFlow = normalizeExternalFlow(raw.externalFlow);
+      const source = { ibkr: "ibkr_flex", schwab: "schwab_transactions", longbridge: "longbridge_cashflow" }[authorizedPlatform];
+      if (externalFlow.windows.some(w => w.source !== source)) throw new Error("external_flow_source_mismatch");
+    } catch (error) {
+      return accountFactsErrorResponse(error);
+    }
+    raw = { ...raw };
+    delete raw.externalFlow;
+  }
   let history;
   try {
     history = normalizeAccountFactsHistoryPayload(raw, { expectedPlatform: authorizedPlatform });
@@ -160,6 +174,7 @@ export async function syncAccountFactsResponse(request, env, deps) {
       target_id: resolved.binding.target_id,
       source_binding_id: resolved.binding.source_binding.id,
       history,
+      ...(externalFlow ? { externalFlow } : {}),
     });
   } catch (error) {
     return accountFactsErrorResponse(error);