/** * Core analysis functions shared between main thread and worker. * All functions here are PURE (no side effects, no external state). * * The clustering algorithm (clusterNewsCore, aggregateThreats, * MAX_CLUSTER_NEWS_ITEMS) and its input/output types now live in * shared/news-clustering-core.js (issue #5697) so server-side MCP tools * cluster identically; they are re-exported here unchanged. This module keeps * the correlation signal detection algorithms, which pull in entity * extraction and other client-coupled modules. * * Both the main-thread services and the Web Worker import from here. */ import { SIMILARITY_THRESHOLD, PREDICTION_SHIFT_THRESHOLD, MARKET_MOVE_THRESHOLD, NEWS_VELOCITY_THRESHOLD, FLOW_PRICE_THRESHOLD, ENERGY_COMMODITY_SYMBOLS, PIPELINE_KEYWORDS, FLOW_DROP_KEYWORDS, TOPIC_KEYWORDS, SUPPRESSED_TRENDING_TERMS, tokenize, jaccardSimilarity, includesKeyword, containsTopicKeyword, findRelatedTopics, generateSignalId, generateDedupeKey, } from '@/utils/analysis-constants'; import { extractEntitiesFromClusters, findNewsForMarketSymbol, } from './entity-extraction'; import { getEntityIndex } from './entity-index'; import { effectivePubDateMs } from './feed-date'; export { MAX_CLUSTER_NEWS_ITEMS, aggregateThreats, clusterNewsCore, } from '../../shared/news-clustering-core.js'; export type { NewsItemCore, NewsItemWithTier, ClusteredEventCore, } from '../../shared/news-clustering-core.js'; import type { ClusteredEventCore } from '../../shared/news-clustering-core.js'; const TOPIC_BASELINE_WINDOW_MS = 7 * 24 * 60 * 60 * 1000; const TOPIC_BASELINE_SPIKE_MULTIPLIER = 3; const TOPIC_HISTORY_MAX_POINTS = 1000; interface TopicVelocityPoint { timestamp: number; velocity: number; } // Re-export for convenience export { SIMILARITY_THRESHOLD, tokenize, jaccardSimilarity, generateSignalId, generateDedupeKey, }; export interface PredictionMarketCore { title: string; yesPrice: number; volume?: number; } export interface MarketDataCore { symbol: string; name: string; display: string; price: number | null; change: number | null; } export type SignalType = | 'prediction_leads_news' | 'news_leads_markets' | 'silent_divergence' | 'velocity_spike' | 'keyword_spike' | 'convergence' | 'triangulation' | 'flow_drop' | 'flow_price_divergence' | 'geo_convergence' | 'explained_market_move' | 'hotspot_escalation' | 'sector_cascade' | 'military_surge'; export interface CorrelationSignalCore { id: string; type: SignalType; title: string; description: string; confidence: number; timestamp: Date; data: { newsVelocity?: number; marketChange?: number; predictionShift?: number; relatedTopics?: string[]; correlatedEntities?: string[]; correlatedNews?: string[]; explanation?: string; term?: string; baseline?: number; multiplier?: number; sourceCount?: number; }; } export type SourceType = 'wire' | 'gov' | 'intel' | 'mainstream' | 'market' | 'tech' | 'other' | 'unknown'; export interface StreamSnapshot { newsVelocity: Map; marketChanges: Map; predictionChanges: Map; topicVelocityHistory: Map; timestamp: number; } // ============================================================================ // CORRELATION FUNCTIONS // ============================================================================ function extractTopics(events: ClusteredEventCore[]): Map { const topics = new Map(); for (const event of events) { const title = event.primaryTitle.toLowerCase(); for (const kw of TOPIC_KEYWORDS) { if (SUPPRESSED_TRENDING_TERMS.has(kw)) continue; if (!containsTopicKeyword(title, kw)) continue; const velocity = event.velocity?.sourcesPerHour ?? 0; topics.set(kw, (topics.get(kw) ?? 0) + velocity + event.sourceCount); } } return topics; } function pruneVelocityHistory(history: TopicVelocityPoint[], now: number): TopicVelocityPoint[] { return history.filter(point => now - point.timestamp <= TOPIC_BASELINE_WINDOW_MS); } function averageVelocity(history: TopicVelocityPoint[]): number { if (history.length === 0) return 0; const total = history.reduce((sum, point) => sum + point.velocity, 0); return total / history.length; } function countRelatedTopicMentions( newsTopics: Map, market: Pick ): number { const marketNameLower = market.name.toLowerCase(); const marketSymbolLower = market.symbol.toLowerCase(); return Array.from(newsTopics.entries()) .filter(([topic]) => marketNameLower.includes(topic) || topic.includes(marketSymbolLower)) .reduce((sum, [, velocity]) => sum + velocity, 0); } export function detectPipelineFlowDrops( events: ClusteredEventCore[], isRecentDuplicate: (key: string) => boolean, markSignalSeen: (key: string) => void ): CorrelationSignalCore[] { const signals: CorrelationSignalCore[] = []; for (const event of events) { const titles = [ event.primaryTitle, ...(event.allItems?.map(item => item.title) ?? []), ] .map(title => title.toLowerCase()) .filter(Boolean); const hasPipeline = titles.some(title => includesKeyword(title, PIPELINE_KEYWORDS)); const hasFlowDrop = titles.some(title => includesKeyword(title, FLOW_DROP_KEYWORDS)); if (hasPipeline && hasFlowDrop) { const dedupeKey = generateDedupeKey('flow_drop', event.id, event.sourceCount); if (!isRecentDuplicate(dedupeKey)) { markSignalSeen(dedupeKey); signals.push({ id: generateSignalId(), type: 'flow_drop', title: 'Pipeline Flow Drop', description: `"${event.primaryTitle.slice(0, 70)}..." indicates reduced flow or disruption`, confidence: Math.min(0.9, 0.4 + event.sourceCount / 10), timestamp: new Date(), data: { newsVelocity: event.sourceCount, relatedTopics: ['pipeline', 'flow'], }, }); } } } return signals; } export function detectConvergence( events: ClusteredEventCore[], getSourceType: (source: string) => SourceType, isRecentDuplicate: (key: string) => boolean, markSignalSeen: (key: string) => void ): CorrelationSignalCore[] { const signals: CorrelationSignalCore[] = []; const WINDOW_MS = 60 * 60 * 1000; const now = Date.now(); for (const event of events) { if (!event.allItems || event.allItems.length < 3) continue; const recentItems = event.allItems.filter( item => now - effectivePubDateMs(item) < WINDOW_MS ); if (recentItems.length < 3) continue; const sourceTypes = new Set(); for (const item of recentItems) { const type = getSourceType(item.source); sourceTypes.add(type); } if (sourceTypes.size >= 3) { const types = Array.from(sourceTypes).filter(t => t !== 'other' && t !== 'unknown'); const dedupeKey = generateDedupeKey('convergence', event.id, sourceTypes.size); if (!isRecentDuplicate(dedupeKey) && types.length >= 3) { markSignalSeen(dedupeKey); signals.push({ id: generateSignalId(), type: 'convergence', title: 'Source Convergence', description: `"${event.primaryTitle.slice(0, 50)}..." reported by ${types.join(', ')} (${recentItems.length} sources in 30m)`, confidence: Math.min(0.95, 0.6 + sourceTypes.size * 0.1), timestamp: new Date(), data: { newsVelocity: recentItems.length, relatedTopics: types, }, }); } } } return signals; } export function detectTriangulation( events: ClusteredEventCore[], getSourceType: (source: string) => SourceType, isRecentDuplicate: (key: string) => boolean, markSignalSeen: (key: string) => void ): CorrelationSignalCore[] { const signals: CorrelationSignalCore[] = []; const CRITICAL_TYPES: SourceType[] = ['wire', 'gov', 'intel']; for (const event of events) { if (!event.allItems || event.allItems.length < 3) continue; const typePresent = new Set(); for (const item of event.allItems) { const t = getSourceType(item.source); if (CRITICAL_TYPES.includes(t)) { typePresent.add(t); } } if (typePresent.size === 3) { const dedupeKey = generateDedupeKey('triangulation', event.id, 3); if (!isRecentDuplicate(dedupeKey)) { markSignalSeen(dedupeKey); signals.push({ id: generateSignalId(), type: 'triangulation', title: 'Intel Triangulation', description: `Wire + Gov + Intel aligned: "${event.primaryTitle.slice(0, 45)}..."`, confidence: 0.9, timestamp: new Date(), data: { newsVelocity: event.sourceCount, relatedTopics: Array.from(typePresent), }, }); } } } return signals; } /** * Analyze correlations between news, predictions, and markets. * Pure function - state management (snapshots, deduplication) handled by caller. */ export function analyzeCorrelationsCore( events: ClusteredEventCore[], predictions: PredictionMarketCore[], markets: MarketDataCore[], previousSnapshot: StreamSnapshot | null, getSourceType: (source: string) => SourceType, isRecentDuplicate: (key: string) => boolean, markSignalSeen: (key: string) => void ): { signals: CorrelationSignalCore[]; snapshot: StreamSnapshot } { const signals: CorrelationSignalCore[] = []; const now = Date.now(); const newsTopics = extractTopics(events); const pipelineFlowSignals = detectPipelineFlowDrops(events, isRecentDuplicate, markSignalSeen); const pipelineFlowMentions = pipelineFlowSignals.length; const entityIndex = getEntityIndex(); const newsEntityContexts = extractEntitiesFromClusters(events); const previousHistory = previousSnapshot?.topicVelocityHistory ?? new Map(); const currentHistory = new Map(); const topicUniverse = new Set([ ...previousHistory.keys(), ...newsTopics.keys(), ]); for (const topic of topicUniverse) { const prior = pruneVelocityHistory(previousHistory.get(topic) ?? [], now); const updated = [...prior, { timestamp: now, velocity: newsTopics.get(topic) ?? 0 }]; if (updated.length > TOPIC_HISTORY_MAX_POINTS) { updated.splice(0, updated.length - TOPIC_HISTORY_MAX_POINTS); } currentHistory.set(topic, updated); } const currentSnapshot: StreamSnapshot = { newsVelocity: newsTopics, marketChanges: new Map(markets.map(m => [m.symbol, m.change ?? 0])), predictionChanges: new Map(predictions.map(p => [p.title.slice(0, 50), p.yesPrice])), topicVelocityHistory: currentHistory, timestamp: now, }; if (!previousSnapshot) { return { signals: [], snapshot: currentSnapshot }; } // Detect prediction shifts for (const pred of predictions) { const key = pred.title.slice(0, 50); const prev = previousSnapshot.predictionChanges.get(key); if (prev !== undefined) { const shift = Math.abs(pred.yesPrice - prev); if (shift >= PREDICTION_SHIFT_THRESHOLD) { const related = findRelatedTopics(pred.title); const newsActivity = related.reduce((sum, t) => sum + (newsTopics.get(t) ?? 0), 0); const dedupeKey = generateDedupeKey('prediction_leads_news', key, shift); if (newsActivity < NEWS_VELOCITY_THRESHOLD && !isRecentDuplicate(dedupeKey)) { markSignalSeen(dedupeKey); signals.push({ id: generateSignalId(), type: 'prediction_leads_news', title: 'Prediction Market Shift', description: `"${pred.title.slice(0, 60)}..." moved ${shift > 0 ? '+' : ''}${shift.toFixed(1)}% with low news coverage`, confidence: Math.min(0.9, 0.5 + shift / 20), timestamp: new Date(), data: { predictionShift: shift, newsVelocity: newsActivity, relatedTopics: related, }, }); } } } } // Detect news velocity spikes for (const [topic, velocity] of newsTopics) { if (SUPPRESSED_TRENDING_TERMS.has(topic)) continue; const baselineHistory = pruneVelocityHistory(previousHistory.get(topic) ?? [], now); const baseline = averageVelocity(baselineHistory); const exceedsAbsoluteThreshold = velocity > NEWS_VELOCITY_THRESHOLD * 2; const exceedsBaseline = baseline > 0 ? velocity > baseline * TOPIC_BASELINE_SPIKE_MULTIPLIER : exceedsAbsoluteThreshold; if (!exceedsAbsoluteThreshold || !exceedsBaseline) continue; const multiplier = baseline > 0 ? velocity / baseline : 0; const dedupeKey = generateDedupeKey('velocity_spike', topic, velocity); if (!isRecentDuplicate(dedupeKey)) { markSignalSeen(dedupeKey); const baselineText = baseline > 0 ? `${baseline.toFixed(1)} baseline (${multiplier.toFixed(1)}x)` : 'cold-start baseline'; signals.push({ id: generateSignalId(), type: 'velocity_spike', title: 'News Velocity Spike', description: `"${topic}" coverage surging: ${velocity.toFixed(1)} activity score vs ${baselineText}`, confidence: Math.min(0.9, 0.45 + (multiplier > 0 ? multiplier / 8 : velocity / 18)), timestamp: new Date(), data: { newsVelocity: velocity, relatedTopics: [topic], baseline, multiplier: baseline > 0 ? multiplier : undefined, explanation: baseline > 0 ? `Velocity ${velocity.toFixed(1)} is ${multiplier.toFixed(1)}x above baseline ${baseline.toFixed(1)}` : `Velocity ${velocity.toFixed(1)} exceeded cold-start threshold`, }, }); } } // Detect market moves with entity-aware news correlation for (const market of markets) { const change = Math.abs(market.change ?? 0); if (change < MARKET_MOVE_THRESHOLD) continue; const entity = entityIndex.byId.get(market.symbol); const relatedNews = findNewsForMarketSymbol(market.symbol, newsEntityContexts); if (relatedNews.length > 0) { const topNews = relatedNews[0]!; const dedupeKey = generateDedupeKey('explained_market_move', market.symbol, change); if (!isRecentDuplicate(dedupeKey)) { markSignalSeen(dedupeKey); const direction = market.change! > 0 ? '+' : ''; signals.push({ id: generateSignalId(), type: 'explained_market_move', title: 'Market Move Explained', description: `${market.name} ${direction}${market.change!.toFixed(2)}% correlates with: "${topNews.title.slice(0, 60)}..."`, confidence: Math.min(0.9, 0.5 + (relatedNews.length * 0.1) + (change / 20)), timestamp: new Date(), data: { marketChange: market.change!, newsVelocity: relatedNews.length, correlatedEntities: [market.symbol], correlatedNews: relatedNews.map(n => n.clusterId), explanation: `${relatedNews.length} related news item${relatedNews.length > 1 ? 's' : ''} found`, }, }); } } else { const oldRelatedNews = countRelatedTopicMentions(newsTopics, market); const dedupeKey = generateDedupeKey('silent_divergence', market.symbol, change); if (oldRelatedNews < 2 && !isRecentDuplicate(dedupeKey)) { markSignalSeen(dedupeKey); const searchedTerms = entity ? [market.symbol, market.name, ...(entity.keywords?.slice(0, 2) ?? [])].join(', ') : market.symbol; signals.push({ id: generateSignalId(), type: 'silent_divergence', title: 'Silent Divergence', description: `${market.name} moved ${market.change! > 0 ? '+' : ''}${market.change!.toFixed(2)}% - no news found for: ${searchedTerms}`, confidence: Math.min(0.8, 0.4 + change / 10), timestamp: new Date(), data: { marketChange: market.change!, newsVelocity: oldRelatedNews, explanation: `Searched: ${searchedTerms}`, }, }); } } } // Detect flow/price divergence for energy commodities for (const market of markets) { if (!ENERGY_COMMODITY_SYMBOLS.has(market.symbol)) continue; const change = market.change ?? 0; if (change >= FLOW_PRICE_THRESHOLD) { const relatedNews = countRelatedTopicMentions(newsTopics, market); const dedupeKey = generateDedupeKey('flow_price_divergence', market.symbol, change); if (relatedNews < 2 && pipelineFlowMentions === 0 && !isRecentDuplicate(dedupeKey)) { markSignalSeen(dedupeKey); signals.push({ id: generateSignalId(), type: 'flow_price_divergence', title: 'Flow/Price Divergence', description: `${market.name} up ${change.toFixed(2)}% without pipeline flow news`, confidence: Math.min(0.85, 0.4 + change / 8), timestamp: new Date(), data: { marketChange: change, newsVelocity: relatedNews, relatedTopics: ['pipeline', market.display], }, }); } } } // Add convergence and triangulation signals signals.push(...detectConvergence(events, getSourceType, isRecentDuplicate, markSignalSeen)); signals.push(...detectTriangulation(events, getSourceType, isRecentDuplicate, markSignalSeen)); signals.push(...pipelineFlowSignals); // Dedupe by type to avoid spam const uniqueSignals = signals.filter((sig, idx) => signals.findIndex(s => s.type === sig.type) === idx ); // Only return high-confidence signals return { signals: uniqueSignals.filter(s => s.confidence >= 0.6), snapshot: currentSnapshot, }; }