investor-flow/app/server/src/services/FinraIngestService.ts

import type { DatabaseSync } from 'node:sqlite';
import { getFinraBaseUrl } from '../admin/admin.ts';

/** Format a FINRA consolidated-short-interest filename: CNMSshvol{YYYYMMDD}.txt */
function finraFilename(settlementDate: string): string {
  const d = settlementDate.replace(/-/g, '');
  return `CNMSshvol${d}.txt`;
}

/** Parse a FINRA CNMS short-vol file body (pipe-delimited).
 *  Actual columns (verified against live file):
 *    Date|Symbol|ShortVolume|ShortExemptVolume|TotalVolume|Market
 *  Where Market is comma-separated exchange codes (B=NYSE, Q=NASDAQ, N=NYSE Amer/ARCA).
 *  Volumes are TOTAL across all listed exchanges — one row per symbol per settlement date. */
function parseFinraFile(
  body: string,
  settlementDate: string,
  ingestedAt: string,
  sourceFile: string,
): Array<{
  symbol: string;
  exchange: string;
  shortVolume: number;
  shortExempt: number;
  totalVolume: number;
}> {
  const lines = body.split(/\r?\n/);
  const rows: Array<{
    symbol: string;
    exchange: string;
    shortVolume: number;
    shortExempt: number;
    totalVolume: number;
  }> = [];
  let headerFound = false;

  for (const raw of lines) {
    const line = raw.trim();
    if (!line || line.startsWith('#')) continue;
    if (line.includes('Date|Symbol|Short') || line.includes('Date|Symbol|')) {
      headerFound = true;
      continue;
    }
    if (!headerFound) continue;

    const cols = line.split('|').map((c) => c.trim());
    if (cols.length < 5) continue;

    // cols[0] = Date (YYYYMMDD), cols[1] = Symbol, cols[4] = TotalVolume, cols[5] = Market
    const symbol = cols[1]?.replace(/\/.*$/, ''); // strip warrant suffix like /WS
    const rawShort = cols[2]?.replace(/,/g, '') ?? '0';
    const rawExempt = cols[3]?.replace(/,/g, '') ?? '0';
    const rawTotal = cols[4]?.replace(/,/g, '') ?? rawShort;
    const markets = cols[5] ?? '';

    const shortVolume = parseFloat(rawShort);
    const shortExempt = parseFloat(rawExempt);
    const totalVolume = parseFloat(rawTotal);

    if (!symbol || Number.isNaN(shortVolume)) continue;

    // Store market codes as combined string (e.g. "B,Q,N")
    const normSymbol = symbol.toUpperCase();
    const exchangeCode = markets.replace(/\s/g, '').replace(/^,+|,+$/g, '') || 'ALL';

    // Dedupe by symbol — the volume is total across all listed exchanges
    if (!rows.some((r) => r.symbol === normSymbol)) {
      rows.push({
        symbol: normSymbol,
        exchange: exchangeCode,
        shortVolume,
        shortExempt,
        totalVolume,
      });
    }
  }
  return rows;
}

/** Download and ingest a FINRA consolidated-short-interest file (CNMSshvol{YYYYMMDD}.txt).
 *  baseUrl resolves from DB config > env FINRA_BASE_URL > built-in default when omitted. */
export async function downloadAndIngestFinra(
  db: DatabaseSync,
  settlementDate: string,
  baseUrl?: string,
): Promise<{ symbolsStored: number; sourceFile: string; exchanges: string[] }> {
  const base = (baseUrl ?? getFinraBaseUrl(db)).replace(/\/+$/, '');
  const filename = finraFilename(settlementDate);
  const url = `${base}/${filename}`;
  const ingestedAt = new Date().toISOString();

  console.log(`[finra] downloading ${url}`);
  const resp = await fetch(url, {
    headers: { 'User-Agent': 'InvestorFlow/1.0 (research) node' },
    signal: AbortSignal.timeout(30_000),
  });
  if (!resp.ok) throw new Error(`FINRA download failed: ${resp.status} ${resp.statusText}`);
  const body = await resp.text();
  if (!body.trim()) throw new Error('FINRA file is empty');

  const rows = parseFinraFile(body, settlementDate, ingestedAt, filename);
  if (!rows.length) throw new Error('No FINRA short interest rows parsed');

  const exchanges = [...new Set(rows.map((r) => r.exchange))];

  const upsert = db.prepare(
    `INSERT OR REPLACE INTO finra_short_interest
       (symbol, settlement_date, exchange, short_volume, short_exempt, total_volume, avg_daily_vol, days_to_cover, source_file, ingested_at)
     VALUES (?, ?, ?, ?, ?, ?, NULL, NULL, ?, ?)`
  );

  db.exec('BEGIN TRANSACTION');
  try {
    for (const r of rows) {
      upsert.run(
        r.symbol,
        settlementDate,
        r.exchange,
        r.shortVolume,
        r.shortExempt,
        r.totalVolume,
        filename,
        ingestedAt,
      );
    }
    db.exec('COMMIT');
  } catch (e) {
    db.exec('ROLLBACK');
    throw e;
  }

  console.log(`[finra] ingested ${rows.length} rows from ${filename} (${exchanges.length} exchanges)`);
  return { symbolsStored: rows.length, sourceFile: filename, exchanges };
}

/** Compute days-to-cover for finra_short_interest rows that have avg_daily_vol set.
 *  Called after avg_daily_vol is populated from external volume data. */
export function computeDaysToCover(db: DatabaseSync): number {
  const r = db.exec(
    `UPDATE finra_short_interest
     SET days_to_cover = CASE
       WHEN avg_daily_vol IS NOT NULL AND avg_daily_vol > 0 THEN short_volume / avg_daily_vol
       ELSE NULL
     END
     WHERE days_to_cover IS NULL AND avg_daily_vol IS NOT NULL`
  );
  return r.changes;
}

/** Get the latest settlement date available in the finra_short_interest table. */
export function latestFinraSettlementDate(db: DatabaseSync): string | null {
  const r = db.prepare(
    'SELECT settlement_date FROM finra_short_interest ORDER BY settlement_date DESC LIMIT 1'
  ).get() as { settlement_date: string } | undefined;
  return r?.settlement_date ?? null;
}