feat(data): extend market data provider with charts, fundamentals, and quant analysis
Adds a stock data cache layer and indicator helpers (MA/RSI/MACD), expands the MarketDataProvider interface with candles, fundamentals, related-data, and quant analysis, and threads tags + source attribution through the shared types. Updates affected agent tests. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
380
web/src/agent/stockDataCache.ts
Normal file
380
web/src/agent/stockDataCache.ts
Normal file
@@ -0,0 +1,380 @@
|
||||
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")}`;
|
||||
}
|
||||
Reference in New Issue
Block a user