quant-trading/web/src/agent/stockDataCache.ts

import { mkdir, readFile, writeFile } from "node:fs/promises";
import { join } from "node:path";
import { calculateIndicators } from "./indicators";
import type {
  ChartPeriod,
  FinancialStatement,
  MarketDataProvider,
  PriceCandle,
  QuantAnalysis,
  RelatedStockData,
  StockChartData,
  StockFundamentalsData,
  StockQuantAnalysis,
  StockRelatedData,
  StockSnapshot,
  WatchStock
} from "../shared/types";

interface CachedEntry<T> {
  cachedAt: string;
  data: T;
}

interface StockCacheFile {
  snapshot?: CachedEntry<StockSnapshot>;
  candles?: CachedEntry<PriceCandle[]>;
  candlesWindowYears?: number;
  candlesProviderVersion?: number;
  fundamentals?: CachedEntry<FinancialStatement[]>;
  relatedData?: CachedEntry<RelatedStockData>;
  quantAnalysis?: CachedEntry<QuantAnalysis> & { version?: number };
}

const SNAPSHOT_TTL_MS = 24 * 60 * 60 * 1000;
const CANDLES_TTL_MS = 24 * 60 * 60 * 1000;
const FUNDAMENTALS_TTL_MS = 7 * 24 * 60 * 60 * 1000;
const RELATED_TTL_MS = 24 * 60 * 60 * 1000;
const ANALYSIS_VERSION = 4;
const CANDLES_WINDOW_YEARS = 10;
const CANDLES_PROVIDER_VERSION = 2;

export class StockDataCache {
  constructor(
    private readonly cacheDir: string,
    private readonly provider: MarketDataProvider,
    private readonly now: () => Date = () => new Date()
  ) {}

  async snapshot(stock: WatchStock, force = false): Promise<StockSnapshot> {
    const cache = await this.read(stock);
    if (!force && cache.snapshot && !isExpired(cache.snapshot.cachedAt, SNAPSHOT_TTL_MS, this.now())) {
      return cache.snapshot.data;
    }

    const next = await this.provider.snapshot(stock);
    await this.write(stock, { ...cache, snapshot: { cachedAt: this.now().toISOString(), data: next } });
    return next;
  }

  async chart(stock: WatchStock, period: ChartPeriod = "D", force = false): Promise<StockChartData> {
    const cache = await this.read(stock);
    const existing = cache.candles;
    let entry: CachedEntry<PriceCandle[]>;
    try {
      if (!existing || force || cache.candlesWindowYears !== CANDLES_WINDOW_YEARS || cache.candlesProviderVersion !== CANDLES_PROVIDER_VERSION) {
        entry = { cachedAt: this.now().toISOString(), data: await this.provider.candles(stock) };
      } else if (isExpired(existing.cachedAt, CANDLES_TTL_MS, this.now())) {
        entry = existing;
        void this.refreshCandles(stock, existing);
      } else {
        entry = existing;
      }
    } catch (error) {
      if (existing) {
        entry = existing;
      } else {
        throw error;
      }
    }
    if (entry !== cache.candles) {
      await this.write(stock, { ...cache, candles: entry, candlesWindowYears: CANDLES_WINDOW_YEARS, candlesProviderVersion: CANDLES_PROVIDER_VERSION });
    }

    const candles = aggregateCandles(entry.data, period);
    return {
      stock,
      period,
      candles,
      indicators: calculateIndicators(candles),
      cachedAt: entry.cachedAt
    };
  }

  private async refreshCandles(stock: WatchStock, existing: CachedEntry<PriceCandle[]>): Promise<void> {
    try {
      const cache = await this.read(stock);
      if (cache.candles?.cachedAt !== existing.cachedAt) {
        return;
      }
      const startDate = previousCalendarDate(existing.data[existing.data.length - 1]?.date);
      const incremental = startDate ? await this.provider.candles(stock, startDate) : await this.provider.candles(stock);
      const entry = {
        cachedAt: this.now().toISOString(),
        data: incremental.length ? mergeCandles(existing.data, incremental) : existing.data
      };
      await this.write(stock, { ...cache, candles: entry, candlesWindowYears: CANDLES_WINDOW_YEARS, candlesProviderVersion: CANDLES_PROVIDER_VERSION });
    } catch {
      // Keep stale cache when the upstream market data source is unavailable.
    }
  }

  async refreshChart(stock: WatchStock, period: ChartPeriod = "D"): Promise<StockChartData> {
    const cache = await this.read(stock);
    const existing = cache.candles;
    let entry: CachedEntry<PriceCandle[]>;
    if (existing && cache.candlesWindowYears === CANDLES_WINDOW_YEARS && cache.candlesProviderVersion === CANDLES_PROVIDER_VERSION) {
        const startDate = previousCalendarDate(existing.data[existing.data.length - 1]?.date);
        const incremental = startDate ? await this.provider.candles(stock, startDate) : await this.provider.candles(stock);
        entry = {
          cachedAt: this.now().toISOString(),
          data: incremental.length ? mergeCandles(existing.data, incremental) : existing.data
        };
    } else {
      entry = { cachedAt: this.now().toISOString(), data: await this.provider.candles(stock) };
    }
    await this.write(stock, { ...cache, candles: entry, candlesWindowYears: CANDLES_WINDOW_YEARS, candlesProviderVersion: CANDLES_PROVIDER_VERSION });

    const candles = aggregateCandles(entry.data, period);
    return {
      stock,
      period,
      candles,
      indicators: calculateIndicators(candles),
      cachedAt: entry.cachedAt
    };
  }

  async fundamentals(stock: WatchStock, force = false): Promise<StockFundamentalsData> {
    const cache = await this.read(stock);
    let entry: CachedEntry<FinancialStatement[]>;
    try {
      entry =
        !force && cache.fundamentals
          ? cache.fundamentals
          : { cachedAt: this.now().toISOString(), data: await this.provider.fundamentals(stock) };
    } catch {
      entry = cache.fundamentals ?? { cachedAt: this.now().toISOString(), data: [] };
    }
    if (entry !== cache.fundamentals) {
      await this.write(stock, { ...cache, fundamentals: entry });
    }

    return {
      stock,
      statements: entry.data,
      cachedAt: entry.cachedAt
    };
  }

  async relatedData(stock: WatchStock, force = false): Promise<StockRelatedData> {
    const cache = await this.read(stock);
    let entry: CachedEntry<RelatedStockData>;
    try {
      entry =
        !force && cache.relatedData && hasResearchSchema(cache.relatedData.data)
          ? cache.relatedData
          : { cachedAt: this.now().toISOString(), data: await this.provider.relatedData(stock) };
    } catch {
      entry = cache.relatedData ?? {
        cachedAt: this.now().toISOString(),
        data: {
          riskFlags: ["相关市场数据源暂不可用。"],
          sources: []
        }
      };
    }
    if (entry !== cache.relatedData) {
      await this.write(stock, { ...cache, relatedData: entry });
    }

    return {
      stock,
      data: entry.data,
      cachedAt: entry.cachedAt
    };
  }

  async quantAnalysis(stock: WatchStock, force = false): Promise<StockQuantAnalysis> {
    const cache = await this.read(stock);
    const hasTodayAnalysis =
      cache.quantAnalysis?.version === ANALYSIS_VERSION && isSameShanghaiDate(cache.quantAnalysis.cachedAt, this.now());
    if (!force && cache.quantAnalysis?.version === ANALYSIS_VERSION) {
      return {
        stock,
        analysis: cache.quantAnalysis.data,
        generatedAt: cache.quantAnalysis.cachedAt,
        cachedAt: cache.quantAnalysis.cachedAt
      };
    }

    const refreshSourceData = force || !hasTodayAnalysis;
    const snapshot = await this.snapshot(stock, refreshSourceData);
    const chart = await this.chart(stock, "D", refreshSourceData);
    const fundamentals = await this.fundamentals(stock, refreshSourceData).catch(() => ({ stock, statements: [], cachedAt: this.now().toISOString() }));
    const related = await this.relatedData(stock, refreshSourceData).catch(() => ({
      stock,
      data: { riskFlags: ["相关市场数据源暂不可用。"], sources: [] },
      cachedAt: this.now().toISOString()
    }));
    const generatedAt = this.now().toISOString();
    const analysis = await this.provider.quantAnalysis({
      stock,
      snapshot,
      candles: chart.candles,
      indicators: chart.indicators,
      fundamentals: fundamentals.statements,
      relatedData: related.data
    });
    const entry = { cachedAt: generatedAt, data: analysis, version: ANALYSIS_VERSION };
    await this.write(stock, { ...(await this.read(stock)), quantAnalysis: entry });

    return {
      stock,
      analysis,
      generatedAt,
      cachedAt: entry.cachedAt
    };
  }

  async analysisNeedsRefresh(stock: WatchStock): Promise<boolean> {
    const cache = await this.read(stock);
    return cache.quantAnalysis?.version !== ANALYSIS_VERSION || !isSameShanghaiDate(cache.quantAnalysis.cachedAt, this.now());
  }

  async hasAnalysis(stock: WatchStock): Promise<boolean> {
    const cache = await this.read(stock);
    return cache.quantAnalysis?.version === ANALYSIS_VERSION;
  }

  placeholderAnalysis(stock: WatchStock): StockQuantAnalysis {
    const generatedAt = this.now().toISOString();
    return {
      stock,
      generatedAt,
      cachedAt: generatedAt,
      analysis: {
        action: "hold",
        score: 50,
        confidence: 0.1,
        horizon: "后台生成中",
        summary: "本地暂无 Agent 分析缓存，已在后台生成最新量化分析。当前先展示缓存数据，避免切换自选股时阻塞界面。",
        factors: [
          { name: "缓存状态", score: 50, weight: 1, evidence: "等待后台 Agent 分析写入本地缓存。" },
          { name: "技术面", score: 50, weight: 0, evidence: "K线和指标可先独立查看。" },
          { name: "风险控制", score: 50, weight: 0, evidence: "完整交易计划将在后台分析完成后更新。" }
        ],
        tradePlan: {
          direction: "wait",
          rationale: "后台分析完成前不生成买入、卖出或止损价。"
        },
        marketContext: ["Agent 分析正在后台生成。"],
        riskNotes: ["该占位结果只表示后台任务状态，不代表实际投资判断。"]
      }
    };
  }

  private async read(stock: WatchStock): Promise<StockCacheFile> {
    try {
      return JSON.parse(await readFile(this.pathFor(stock), "utf8")) as StockCacheFile;
    } catch (error) {
      if (isMissingFile(error)) {
        return {};
      }
      throw error;
    }
  }

  private async write(stock: WatchStock, cache: StockCacheFile): Promise<void> {
    await mkdir(this.cacheDir, { recursive: true });
    await writeFile(this.pathFor(stock), `${JSON.stringify(cache, null, 2)}\n`, "utf8");
  }

  private pathFor(stock: WatchStock): string {
    return join(this.cacheDir, `${stock.display.replace(/[^A-Z0-9]/g, "_")}.json`);
  }
}

function isExpired(cachedAt: string, ttlMs: number, now: Date): boolean {
  const cachedTime = new Date(cachedAt).getTime();
  return !Number.isFinite(cachedTime) || now.getTime() - cachedTime > ttlMs;
}

function isMissingFile(error: unknown): boolean {
  return typeof error === "object" && error !== null && "code" in error && error.code === "ENOENT";
}

function mergeCandles(existing: PriceCandle[], incoming: PriceCandle[]): PriceCandle[] {
  const byDate = new Map<string, PriceCandle>();
  for (const candle of existing) {
    byDate.set(candle.date, candle);
  }
  for (const candle of incoming) {
    byDate.set(candle.date, candle);
  }
  return Array.from(byDate.values()).sort((left, right) => left.date.localeCompare(right.date));
}

function previousCalendarDate(date: string | undefined): string | undefined {
  if (!date) {
    return undefined;
  }
  const value = new Date(`${date}T00:00:00.000Z`);
  if (!Number.isFinite(value.getTime())) {
    return undefined;
  }
  value.setUTCDate(value.getUTCDate() - 7);
  return value.toISOString().slice(0, 10);
}

function hasResearchSchema(data: RelatedStockData): boolean {
  return Array.isArray(data.sources) && data.sources.some((source) => source.name.startsWith("Nasdaq") || source.name.startsWith("Sina"));
}

function isSameShanghaiDate(cachedAt: string, now: Date): boolean {
  const cachedTime = new Date(cachedAt);
  if (!Number.isFinite(cachedTime.getTime())) {
    return false;
  }
  return shanghaiDate(cachedTime) === shanghaiDate(now);
}

function shanghaiDate(date: Date): string {
  return new Intl.DateTimeFormat("en-CA", {
    timeZone: "Asia/Shanghai",
    year: "numeric",
    month: "2-digit",
    day: "2-digit"
  }).format(date);
}

function aggregateCandles(candles: PriceCandle[], period: ChartPeriod): PriceCandle[] {
  if (period === "D") {
    return candles;
  }

  const grouped = new Map<string, PriceCandle>();
  for (const candle of candles) {
    const key = periodKey(candle.date, period);
    const existing = grouped.get(key);
    if (!existing) {
      grouped.set(key, { ...candle });
      continue;
    }
    existing.high = Math.max(existing.high, candle.high);
    existing.low = Math.min(existing.low, candle.low);
    existing.close = candle.close;
    existing.volume += candle.volume;
    existing.date = candle.date;
  }
  return Array.from(grouped.values());
}

function periodKey(date: string, period: ChartPeriod): string {
  if (period === "M") {
    return date.slice(0, 7);
  }
  if (period === "Y") {
    return date.slice(0, 4);
  }
  return isoWeekKey(date);
}

function isoWeekKey(date: string): string {
  const value = new Date(`${date}T00:00:00.000Z`);
  const day = value.getUTCDay() || 7;
  value.setUTCDate(value.getUTCDate() + 4 - day);
  const yearStart = new Date(Date.UTC(value.getUTCFullYear(), 0, 1));
  const week = Math.ceil(((value.getTime() - yearStart.getTime()) / 86400000 + 1) / 7);
  return `${value.getUTCFullYear()}-W${String(week).padStart(2, "0")}`;
}