const https = require("https"); const http = require("http"); const { findMatchedCompaniesByEmbedding } = require("./embeddings"); const { getPriceContext, formatPriceContext } = require("./priceContext"); const REPROCESS_MIN_NEW_ARTICLES = 3; const REPROCESS_COOLDOWN_HOURS = 6; async function runAugorWorker(archiveDb, intelligenceDb, config) { const loopDelay = config.workers?.augorLoopDelayMs ?? 1500; const llmConfig = config.openRouter || {}; const getPending = intelligenceDb.prepare(` SELECT * FROM article_queue WHERE status = 'pending' LIMIT 1 `); const getProcessingState = intelligenceDb.prepare( "SELECT last_processed_at, articles_at_last_run FROM event_processing_state WHERE event_id = ?" ); const upsertProcessingState = intelligenceDb.prepare(` INSERT INTO event_processing_state (event_id, last_processed_at, articles_at_last_run) VALUES (?, CURRENT_TIMESTAMP, ?) ON CONFLICT(event_id) DO UPDATE SET last_processed_at = CURRENT_TIMESTAMP, articles_at_last_run = excluded.articles_at_last_run `); const getCompanyAccuracy = intelligenceDb.prepare(` SELECT COUNT(*) as total, SUM(correct_10d) as correct FROM prediction_outcomes WHERE company_id = ? AND correct_10d IS NOT NULL `); const recordEvent = intelligenceDb.prepare( `INSERT INTO worker_events (worker) VALUES ('augor')` ); const pruneEvents = intelligenceDb.prepare( `DELETE FROM worker_events WHERE worker = 'augor' AND completed_at < datetime('now', '-1 hour')` ); let pruneCounter = 0; const setStatus = intelligenceDb.prepare(` UPDATE article_queue SET status = ?, updated_at = CURRENT_TIMESTAMP WHERE article_id = ? `); const getEventArticleIds = archiveDb.prepare( "SELECT id FROM articles WHERE event_id = ?" ); const setStatusByArticleId = intelligenceDb.prepare(` UPDATE article_queue SET status = 'processed', updated_at = CURRENT_TIMESTAMP WHERE article_id = ? AND status = 'pending' `); const deleteKnowledge = intelligenceDb.prepare( "DELETE FROM event_knowledge WHERE event_id = ?" ); const deletePredictions = intelligenceDb.prepare( "DELETE FROM event_predictions WHERE event_id = ?" ); const insertKnowledge = intelligenceDb.prepare(` INSERT INTO event_knowledge (event_id, company_id, type, data, event_date) VALUES (?, ?, ?, ?, ?) `); const insertPrediction = intelligenceDb.prepare(` INSERT INTO event_predictions (event_id, company_id, type, direction, magnitude, timeframe, rationale, probability, event_date) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) `); const getEventDate = archiveDb.prepare(` SELECT pub_date_effective FROM articles WHERE event_id = ? AND pub_date_effective IS NOT NULL ORDER BY pub_date_effective ASC LIMIT 1 `); const getCompanyFacts = intelligenceDb.prepare(` SELECT claim, confidence, confirmation_count FROM company_facts WHERE company_id = ? ORDER BY confirmation_count DESC LIMIT 30 `); while (true) { try { const queueRow = getPending.get(); if (!queueRow) { continue; } const article = archiveDb.prepare(` SELECT id, event_id, content, has_embedding FROM articles WHERE id = ? `).get(queueRow.article_id); if (!article || !article.content || !article.has_embedding || !article.event_id) { setStatus.run("skipped", queueRow.article_id); continue; } const eventId = article.event_id; const event = archiveDb.prepare("SELECT * FROM events WHERE id = ?").get(eventId); if (!event) { setStatus.run("skipped", queueRow.article_id); continue; } const eventArticles = archiveDb.prepare(` SELECT id, title, description, content FROM articles WHERE event_id = ? AND content IS NOT NULL AND content != '' ORDER BY id ASC LIMIT 25 `).all(eventId); const eventArticleIds = eventArticles.map(a => a.id); // event-level batching guard — skip re-processing if we already ran on this event recently // and not enough new articles have arrived to justify another LLM call const state = getProcessingState.get(eventId); if (state && state.last_processed_at) { const newArticles = eventArticles.length - state.articles_at_last_run; const lastRunMs = new Date(state.last_processed_at + "Z").getTime(); const hoursSince = (Date.now() - lastRunMs) / 3_600_000; if (newArticles < REPROCESS_MIN_NEW_ARTICLES && hoursSince < REPROCESS_COOLDOWN_HOURS) { for (const r of getEventArticleIds.all(eventId)) setStatusByArticleId.run(r.id); console.log(`[augor] event ${eventId} — only ${newArticles} new articles in ${hoursSince.toFixed(1)}h, skipping re-process`); continue; } } const matchedCompanies = findMatchedCompaniesByEmbedding( eventArticleIds, archiveDb, intelligenceDb, config ); if (matchedCompanies.length === 0) { for (const r of getEventArticleIds.all(eventId)) setStatusByArticleId.run(r.id); upsertProcessingState.run(eventId, eventArticles.length); console.log(`[augor] event ${eventId} — no company match, skipped`); continue; } deleteKnowledge.run(eventId); deletePredictions.run(eventId); const eventDateRow = getEventDate.get(eventId); const eventDate = eventDateRow ? eventDateRow.pub_date_effective : null; const eventDateOnly = eventDate ? eventDate.slice(0, 10) : null; const articleText = eventArticles.map((a, i) => { const body = (a.content || a.description || "").slice(0, 2000); return `[Article ${i + 1}] ${a.title}\n${body}`; }).join("\n\n---\n\n"); for (const company of matchedCompanies) { try { const facts = getCompanyFacts.all(company.id); let factsBlock = null; if (facts.length > 0) { const lines = facts.map(f => `- ${f.claim} (confirmed ${f.confirmation_count} times)`).join("\n"); factsBlock = `Known facts about ${company.name}:\n${lines}`; } // pull live market context for the company at the time of the event let priceBlock = null; if (company.ticker && eventDateOnly) { try { const snapshot = await getPriceContext(intelligenceDb, company.ticker, eventDateOnly); priceBlock = formatPriceContext(snapshot, company.ticker); } catch (_) {} } // historical accuracy of past predictions for this company let accuracyBlock = null; const acc = getCompanyAccuracy.get(company.id); if (acc && acc.total >= 5) { const pct = (acc.correct / acc.total * 100).toFixed(0); accuracyBlock = `Past prediction accuracy for ${company.name}: ${pct}% over ${acc.total} evaluated calls.`; } const result = await callLlm(llmConfig, buildPrompt(company.name, event.title, articleText, factsBlock, priceBlock, accuracyBlock)); if (result) { const seenPreds = new Set(); const writeAll = intelligenceDb.transaction(() => { for (const r of (result.knowledge?.relationships || [])) { insertKnowledge.run(eventId, company.id, "relationship", JSON.stringify(r), eventDate); } for (const t of (result.knowledge?.themes || [])) { insertKnowledge.run(eventId, company.id, "theme", JSON.stringify(t), eventDate); } for (const f of (result.knowledge?.factors || [])) { insertKnowledge.run(eventId, company.id, "factor", JSON.stringify(f), eventDate); } for (const p of (result.predictions || [])) { // hard filter — neutral predictions are dead weight, skip them if (p.direction === "neutral") continue; if (p.direction !== "positive" && p.direction !== "negative") continue; const key = `${p.type}|${p.direction}|${p.magnitude}|${p.timeframe}`; if (seenPreds.has(key)) continue; seenPreds.add(key); const prob = typeof p.probability === "number" && p.probability >= 0 && p.probability <= 1 ? p.probability : null; insertPrediction.run( eventId, company.id, p.type, p.direction, p.magnitude, p.timeframe, p.rationale, prob, eventDate ); } }); writeAll(); } } catch (llmErr) { console.error(`[augor] LLM error for ${company.name} on event ${eventId}:`, llmErr.message); } } for (const r of getEventArticleIds.all(eventId)) setStatusByArticleId.run(r.id); upsertProcessingState.run(eventId, eventArticles.length); recordEvent.run(); pruneCounter++; if (pruneCounter >= 100) { pruneEvents.run(); pruneCounter = 0; } console.log(`[augor] processed event ${eventId} (${matchedCompanies.length} companies, ${eventArticles.length} articles)`); } catch (err) { console.error("[augor] error:", err.message); } finally { // Enforce pacing on every path, including the many early `continue` // branches for skipped or already-processed events. await sleep(loopDelay); } } } function buildPrompt(companyName, eventTitle, articleText, factsBlock, priceBlock, accuracyBlock) { const factsPart = factsBlock ? `${factsBlock}\n\n` : ""; const pricePart = priceBlock ? `Market context for ${companyName}:\n${priceBlock}\n\n` : ""; const accPart = accuracyBlock ? `${accuracyBlock}\n\n` : ""; return `You are a financial intelligence analyst. Always respond in English. ${factsPart}${pricePart}${accPart}Event: ${eventTitle} ${articleText} Analyze the impact of this event on ${companyName}. Return JSON only — no explanation. Strict rules — read carefully: - predictions must be directly caused by THIS specific event — no speculation, no priced-in narrative - only emit a prediction if the evidence is unambiguous AND the effect on ${companyName} is concrete and quantifiable - the default answer is no prediction. an empty predictions array is the correct output for most news. only emit one when the event clearly moves the needle - never emit two predictions that share type+direction+magnitude+timeframe — collapse them into one - direction must be "positive" or "negative" only — no neutral predictions, ever. if the impact is unclear, emit nothing - magnitude: "high" = major revenue/market-share shift, expected >5% stock move "medium" = measurable but limited, expected 1-5% stock move omit any prediction that doesnt clear the medium bar - timeframe: "short" = days to 2 weeks (use sparingly — short-horizon predictions are unreliable) "medium" = 2 weeks to 3 months "long" = 3+ months — preferred when the thesis is structural - probability: your honest calibrated probability that the directional call is correct over the stated timeframe, as a number between 0.5 and 0.95. if you cant honestly assign >= 0.6, dont emit the prediction - if the market context shows the stock has already moved sharply (>10% in 30 days), be sceptical that this event adds new information — the move may already be priced in { "knowledge": { "relationships": [ { "type": "supplier|customer|competitor", "entity": "string", "confidence": "high|medium|low", "evidence": "string" } ], "themes": [ { "theme": "string", "direction": "increasing|stable|decreasing", "evidence": "string" } ], "factors": [ { "factor": "string", "relationship": "string", "evidence": "string" } ] }, "predictions": [ { "type": "market_share|stock_price|competitive_position|other", "direction": "positive|negative", "magnitude": "high|medium", "timeframe": "short|medium|long", "probability": 0.0, "rationale": "string" } ] } Only include claims directly supported by the articles. Use empty arrays if nothing applies.`; } async function callLlm(llmConfig, prompt) { const body = JSON.stringify({ model: llmConfig.llmModel || llmConfig.model, messages: [{ role: "user", content: prompt }], temperature: 0.1, }); const url = new URL("https://openrouter.ai/api/v1/chat/completions"); const responseText = await httpPost(url, body, { "Content-Type": "application/json", "Authorization": `Bearer ${llmConfig.apiKey || ""}`, }); let parsed; try { parsed = JSON.parse(responseText); } catch (e) { throw new Error(`LLM response not JSON: ${responseText.slice(0, 300)}`); } const content = parsed.choices?.[0]?.message?.content; if (!content) return null; const stripped = content.replace(/^```(?:json)?\s*/i, '').replace(/\s*```$/, '').trim(); return JSON.parse(stripped); } function httpPost(url, body, headers) { return new Promise((resolve, reject) => { const lib = url.protocol === "https:" ? https : http; const req = lib.request({ hostname: url.hostname, port: url.port || (url.protocol === "https:" ? 443 : 80), path: url.pathname + url.search, method: "POST", headers: { ...headers, "Content-Length": Buffer.byteLength(body) }, }, (res) => { let data = ""; res.on("data", chunk => data += chunk); res.on("end", () => { if (res.statusCode >= 200 && res.statusCode < 300) { resolve(data); } else { reject(new Error(`LLM ${res.statusCode}: ${data.slice(0, 300)}`)); } }); }); req.on("error", reject); req.write(body); req.end(); }); } function sleep(ms) { return new Promise(r => setTimeout(r, ms)); } module.exports = { runAugorWorker };