one_divine_lot/scripts/test-quote-sync.mjs

/**
 * QuoteSync / QuoteHub 回归测试（R-015 / 迭代 13）
 *
 * 运行：node scripts/test-quote-sync.mjs
 * 隔离：纯内存 mock（tick/instrument/positions 可编程）；DROP 幂等用临时目录真实 SQLite 验证。
 */

import { mkdtempSync } from 'node:fs';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { DatabaseSync } from 'node:sqlite';

let passed = 0;
let failed = 0;
function ok(cond, name) {
  if (cond) { passed++; console.log('  \u2713 ' + name); }
  else { failed++; console.error('  \u2717 ' + name); }
}

function pos(code) { return { code, name: 's-' + code }; }
function tick(code, price) { return { [code]: { lastPrice: price, lastClose: price - 0.1, open: price - 0.05, high: price + 0.02, low: price - 0.08, volume: 100, amount: 1000, time: 1, timetag: 'x', updatedAt: 1 } }; }

function mockDataSource({ tradingDay = '20260902' } = {}) {
  const ds = {
    name: 'mock',
    _positions: [pos('600519.SH'), pos('300057.SZ')],
    _ticks: { '600519.SH': 10, '300057.SZ': 5 },
    _instruments: { '600519.SH': { upStopPrice: 11, downStopPrice: 9, preClose: 9 } },
    _tradingDay: tradingDay,
    _failTicks: false,
    _tickCalls: 0,
    _instrumentCalls: 0,
    async getPositions() { return ds._positions; },
    async getTicks(codes) {
      ds._tickCalls++;
      if (ds._failTicks) throw new Error('tick boom');
      const o = {};
      for (const c of codes) {
        if (ds._ticks[c] != null) o[c] = { lastPrice: ds._ticks[c], lastClose: ds._ticks[c] - 0.1, open: 1, high: 2, low: 0.5, volume: 1, amount: 1, time: 1, timetag: 'x', updatedAt: Date.now() };
      }
      return o;
    },
    async getAsset() { return { accountId: 'A', tradingDate: ds._tradingDay }; },
    async getInstrument(code) {
      ds._instrumentCalls++;
      const i = ds._instruments[code];
      if (!i) throw new Error('no instrument ' + code);
      return { code, ...i, tradingDay: ds._tradingDay };
    },
    async isAvailable() { return true; },
  };
  return ds;
}

const { QuoteHub } = await import('../src/market/QuoteHub.js');
const { QuoteSync } = await import('../src/market/QuoteSync.js');

console.log('[1] prime + 定时同步基本流');
{
  const ds = mockDataSource();
  const hub = new QuoteHub({});
  const sync = new QuoteSync({ hub, runtime: { settings: {}, dataSource: ds }, intervalMs: 30 });
  sync.start();
  await new Promise((r) => setTimeout(r, 120));
  const base = sync.stats.syncCount; // prime 完成时定时轮可能已跑若干次，以此为基线
  sync.stop();
  ok(hub.quotes.size === 2, 'prime 后快照 2 只');
  ok(hub.watchCodes.size === 2, 'watch 集合由 prime 登记');
  ok(sync.stats.syncCount >= 1, '定时轮已运行（' + sync.stats.syncCount + ' 轮）');
}

console.log('[2] 同步失败保留上次价');
{
  const ds = mockDataSource();
  const hub = new QuoteHub({});
  const sync = new QuoteSync({ hub, runtime: { settings: {}, dataSource: ds }, intervalMs: 20 });
  sync.start();
  await new Promise((r) => setTimeout(r, 80));
  sync.stop();
  const before = hub.quotes.get('600519.SH').lastPrice;
  ds._failTicks = true;
  const r = await sync.syncNow();
  ok(r === null, '失败返回 null');
  ok(hub.quotes.get('600519.SH').lastPrice === before, '内存价格未被动（保留上次价）');
  ok(sync.stats.failCount >= 1 && sync.stats.lastError.includes('tick boom'), '失败统计与错误记录（累计语义）');
  ds._failTicks = false;
  await sync.syncNow();
  ok(sync.getStatus().failCount === 0 || hub.quotes.get('600519.SH').lastPrice > 0, '恢复后快照可用');
}

console.log('[3] watch 空不空转 + 读穿透回填');
{
  const ds = mockDataSource();
  const hub = new QuoteHub({});
  const sync = new QuoteSync({ hub, runtime: { settings: {}, dataSource: ds }, intervalMs: 20 });
  sync.start();
  await new Promise((r) => setTimeout(r, 60));
  sync.stop();
  hub.quotes.clear(); // 模拟重启后内存空（watch 仍在）
  const callsBefore = ds._tickCalls;
  const out = await hub.getQuotes(['600519.SH'], { dataSource: ds });
  ok(out['600519.SH']?.lastPrice === 10, '读穿透命中（内存 miss → getTicks 回填）');
  ok(hub.quotes.has('600519.SH'), '读穿透结果回填内存');
  const c1 = ds._tickCalls - callsBefore;
  await hub.getQuotes(['600519.SH'], { dataSource: ds });
  ok(ds._tickCalls - callsBefore === c1, '第二次读走内存不再穿透');
  // watch 空：syncNow 不打 tick
  const hub2 = new QuoteHub({});
  const sync2 = new QuoteSync({ hub: hub2, runtime: { settings: {}, dataSource: ds }, intervalMs: 20 });
  const calls0 = ds._tickCalls;
  await sync2.syncNow();
  ok(ds._tickCalls === calls0, 'watch 空 → 不空转');
}

console.log('[4] 涨停/跌停按交易日缓存 + 日切失效');
{
  const ds = mockDataSource();
  const hub = new QuoteHub({});
  const sync = new QuoteSync({ hub, runtime: { settings: {}, dataSource: ds }, intervalMs: 20 });
  sync.start();
  await new Promise((r) => setTimeout(r, 80));
  sync.stop();
  const calls0 = ds._instrumentCalls;
  const i1 = await hub.getInstrumentCached('600519.SH', '20260902', ds.getInstrument);
  ok(i1.upStopPrice === 11 && i1.downStopPrice === 9, '合约信息返回涨停/跌停');
  const c0 = ds._instrumentCalls;
  await hub.getInstrumentCached('600519.SH', '20260902', ds.getInstrument);
  ok(ds._instrumentCalls === c0, '同交易日命中缓存不重拉');
  const i2 = await hub.getInstrumentCached('600519.SH', '20260903', ds.getInstrument);
  ok(i2 && ds._instrumentCalls > c0, '日切 → 失效重拉');
  ok(!hub._instrumentFresh('300057.SZ', '20260902'), '无缓存的 code 判定为失效');
}

console.log('[5] 热切换清合约缓存');
{
  const ds = mockDataSource();
  const hub = new QuoteHub({});
  const sync = new QuoteSync({ hub, runtime: { settings: {}, dataSource: ds }, intervalMs: 100000 });
  sync.start();
  await new Promise((r) => setTimeout(r, 60));
  await hub.getInstrumentCached('600519.SH', '20260902', ds.getInstrument);
  ok(hub.instruments.size === 1, '合约缓存已建立');
  sync.setBaseUrl('http://new-host:8610');
  ok(hub.instruments.size === 0, '热切换 → 合约缓存清空');
  sync.stop();
}

console.log('[6] ingest 来源无关 + watch 过滤保留');
{
  const hub = new QuoteHub({});
  hub.watch(['A.SH']);
  hub.ingest({ 'A.SH': { lastPrice: 1 }, 'B.SH': { lastPrice: 2 } }, { source: 'rest' });
  ok(hub.quotes.has('A.SH') && !hub.quotes.has('B.SH'), 'watch 过滤生效（未关注的 code 不入快照）');
  hub.ingest({ 'A.SH': { lastPrice: 3 } }, { source: 'future-ws' });
  ok(hub.quotes.get('A.SH').lastPrice === 3, 'source 无关，增量合并');
}

console.log('[7] DROP 幂等（旧结构库 → init 后行情表消失、持仓无损）');
{
  const dir = mkdtempSync(join(tmpdir(), 'odl-quote-drop-'));
  const db = new DatabaseSync(join(dir, 'store.db'));
  db.exec("CREATE TABLE market_quotes_cache (code TEXT PRIMARY KEY, last_price REAL, last_close REAL, updated_at INTEGER); CREATE TABLE strategy_holdings (holding_id INTEGER PRIMARY KEY AUTOINCREMENT, strategy_id TEXT, code TEXT, shares REAL, created_at INTEGER, closed_at INTEGER); INSERT INTO strategy_holdings (strategy_id, code, shares, created_at, closed_at) VALUES ('g', '600519.SH', 100, 1, NULL);");
  db.close();
  const { SqliteStore } = await import('../src/storage/SqliteStore.js');
  const store = new SqliteStore({ dataDir: dir });
  store.init();
  const tables = store.db.prepare("SELECT name FROM sqlite_master WHERE type='table'").all().map((r) => r.name);
  ok(!tables.includes('market_quotes_cache'), '旧行情表被 DROP');
  ok(store.getCurrentHoldings().length === 1, '持仓数据无损');
  store.init();
  ok(true, '二次 init 幂等');
  store.close();
}

console.log('[8] sync-status 数据形态（state 推导边界）');
{
  const now = Date.now();
  const judge = (status) => (!status || !status.syncedAt ? 'never' : now - status.syncedAt <= status.periodMs * 3 ? 'fresh' : 'stale');
  ok(judge({ syncedAt: now - 5000, periodMs: 10000 }) === 'fresh', '10s 周期 5s 前 → fresh');
  ok(judge({ syncedAt: now - 31000, periodMs: 10000 }) === 'stale', '10s 周期 31s 前 → stale');
  ok(judge(null) === 'never', '从未同步 → never');
  ok(judge({ syncedAt: now - 4000, periodMs: 5000 }) === 'fresh', '5s 周期 4s 前 → fresh');
  ok(judge({ syncedAt: now - 16000, periodMs: 5000 }) === 'stale', '5s 周期 16s 前 → stale');
}

console.log('RESULT: passed=' + passed + ' failed=' + failed);
process.exit(failed > 0 ? 1 : 0);