add event_date column to event_knowledge and event_predictions tables; update related logic in admin panel and augorWorker

This commit is contained in:
ImBenji 2026-04-23 14:01:02 +01:00
parent 29c6cdc6c3
commit 7830ca7221
7 changed files with 1077 additions and 177 deletions

File diff suppressed because it is too large Load diff

View file

@ -46,6 +46,12 @@ async function runAugorWorker(archiveDb, intelligenceDb, config) {
ORDER BY pub_date_effective ASC LIMIT 1 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) { while (true) {
try { try {
const queueRow = getPending.get(); const queueRow = getPending.get();
@ -107,7 +113,15 @@ async function runAugorWorker(archiveDb, intelligenceDb, config) {
for (const company of matchedCompanies) { for (const company of matchedCompanies) {
try { try {
const result = await callLlm(llmConfig, buildPrompt(company.name, event.title, articleText)); 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}`;
}
const result = await callLlm(llmConfig, buildPrompt(company.name, event.title, articleText, factsBlock));
if (result) { if (result) {
const writeAll = intelligenceDb.transaction(() => { const writeAll = intelligenceDb.transaction(() => {
@ -144,8 +158,12 @@ async function runAugorWorker(archiveDb, intelligenceDb, config) {
} }
} }
function buildPrompt(companyName, eventTitle, articleText) { function buildPrompt(companyName, eventTitle, articleText, factsBlock) {
return `You are a financial intelligence analyst. Extract structured knowledge about ${companyName} from these news articles. const factsPart = factsBlock ? `${factsBlock}\n\n` : "";
return `You are a financial intelligence analyst focused on ${companyName}. Always respond in English regardless of the language of the input articles.
${factsPart}Assess the impact of the following news event on ${companyName} given what you already know about the company.
Event: ${eventTitle} Event: ${eventTitle}

View file

@ -176,7 +176,7 @@ async function normalizeClaims(companyName, rawClaims, llmConfig) {
const lines = capped.map((c, i) => `${i + 1}. [${c.type}] ${c.text}`).join("\n"); const lines = capped.map((c, i) => `${i + 1}. [${c.type}] ${c.text}`).join("\n");
const prompt = `You are normalizing knowledge claims about ${companyName}. const prompt = `You are normalizing knowledge claims about ${companyName}. Always respond in English regardless of the language of the input articles.
Group semantically equivalent claims and provide a canonical short form for each group (lowercase, no punctuation). Group semantically equivalent claims and provide a canonical short form for each group (lowercase, no punctuation).

View file

@ -87,6 +87,22 @@ function runMigrations(db) {
CREATE UNIQUE INDEX IF NOT EXISTS idx_company_facts_unique ON company_facts (company_id, claim); CREATE UNIQUE INDEX IF NOT EXISTS idx_company_facts_unique ON company_facts (company_id, claim);
CREATE TABLE IF NOT EXISTS company_relationships (
id INTEGER PRIMARY KEY,
from_company_id INTEGER,
relationship_type TEXT,
to_entity TEXT,
to_company_id INTEGER,
confidence TEXT,
confirmation_count INTEGER DEFAULT 1,
first_seen_at DATETIME,
last_seen_at DATETIME,
supporting_event_ids TEXT
);
CREATE UNIQUE INDEX IF NOT EXISTS idx_company_relationships_unique
ON company_relationships (from_company_id, relationship_type, to_entity);
`); `);
} }

258
intelligence/graphWorker.js Normal file
View file

@ -0,0 +1,258 @@
// builds cross-company relationship graph from company_facts rows where type = 'relationship'
const VALID_TYPES = ["supplier", "customer", "competitor", "partner", "investor", "dependency"];
const KEYWORD_MAP = [
["manufactur", "supplier"],
["suppli", "supplier"],
["distribut", "supplier"],
["compet", "competitor"],
["rival", "competitor"],
["invest", "investor"],
["fund", "investor"],
["backer", "investor"],
["customer", "customer"],
["client", "customer"],
["depend", "dependency"],
["partner", "partner"],
["collaborat", "partner"],
["joint", "partner"],
];
const RECIPROCAL = {
supplier: "customer",
customer: "supplier",
competitor: "competitor",
partner: "partner",
investor: "dependency",
dependency: "investor",
};
function parseRelationshipType(rawType) {
const t = (rawType || "").toLowerCase().trim();
if (VALID_TYPES.includes(t)) return t;
for (const [kw, mapped] of KEYWORD_MAP) {
if (t.includes(kw)) return mapped;
}
return "partner";
}
function parseClaim(claim) {
// format from consolidationWorker: "entity is a type"
const parts = claim.split(" is a ");
if (parts.length < 2) return null;
const toEntity = parts[0].trim();
const rawType = parts.slice(1).join(" is a ").trim();
if (!toEntity || !rawType) return null;
return {
toEntity,
relationshipType: parseRelationshipType(rawType),
};
}
function resolveCompany(toEntity, companies) {
const needle = toEntity.toLowerCase();
for (const co of companies) {
if (co.name.toLowerCase() === needle) return co;
if (co.ticker && co.ticker.toLowerCase() === needle) return co;
let aliases = [];
try { aliases = JSON.parse(co.aliases || "[]"); } catch (_) {}
for (const alias of aliases) {
if (alias.toLowerCase() === needle) return co;
}
}
return null;
}
function confidenceFromCount(count) {
if (count >= 8) return "very_high";
if (count >= 4) return "high";
if (count >= 2) return "medium";
return "low";
}
function mergeEventIds(existingJson, newJson) {
let existing = [];
let incoming = [];
try { existing = JSON.parse(existingJson || "[]"); } catch (_) {}
try { incoming = JSON.parse(newJson || "[]"); } catch (_) {}
const merged = [...new Set([...existing, ...incoming])];
return JSON.stringify(merged);
}
async function runGraphWorker(archiveDb, intelligenceDb, config) {
const loopDelay = config.workers?.graphWorkerLoopDelayMs ?? 90000;
const getRelationshipFacts = intelligenceDb.prepare(`
SELECT * FROM company_facts WHERE type = 'relationship'
`);
const getCompanies = intelligenceDb.prepare("SELECT * FROM tracked_companies");
const getCompanyById = intelligenceDb.prepare("SELECT * FROM tracked_companies WHERE id = ?");
const getExistingRel = intelligenceDb.prepare(`
SELECT * FROM company_relationships
WHERE from_company_id = ? AND relationship_type = ? AND to_entity = ?
`);
const insertRel = intelligenceDb.prepare(`
INSERT INTO company_relationships
(from_company_id, relationship_type, to_entity, to_company_id, confidence, confirmation_count, first_seen_at, last_seen_at, supporting_event_ids)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
`);
const updateRel = intelligenceDb.prepare(`
UPDATE company_relationships
SET confirmation_count = ?,
confidence = ?,
last_seen_at = ?,
supporting_event_ids = ?
WHERE from_company_id = ? AND relationship_type = ? AND to_entity = ?
`);
const getExistingReciprocal = intelligenceDb.prepare(`
SELECT id FROM company_relationships
WHERE from_company_id = ? AND relationship_type = ? AND to_entity = ?
`);
const insertReciprocal = intelligenceDb.prepare(`
INSERT OR IGNORE INTO company_relationships
(from_company_id, relationship_type, to_entity, to_company_id, confidence, confirmation_count, first_seen_at, last_seen_at, supporting_event_ids)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
`);
while (true) {
try {
const facts = getRelationshipFacts.all();
const companies = getCompanies.all();
let upserted = 0;
let reciprocals = 0;
const processAll = intelligenceDb.transaction(() => {
for (const fact of facts) {
const parsed = parseClaim(fact.claim);
if (!parsed) continue;
const { toEntity, relationshipType } = parsed;
const resolved = resolveCompany(toEntity, companies);
const toCompanyId = resolved ? resolved.id : null;
const existing = getExistingRel.get(fact.company_id, relationshipType, toEntity);
const now = new Date().toISOString();
let finalCount;
let finalFirst;
let finalLast;
let finalEventIds;
if (existing) {
// derive count from the fact itself, not accumulate — avoids compounding on re-runs
finalCount = fact.confirmation_count;
finalFirst = existing.first_seen_at;
finalLast = fact.last_seen_at || now;
finalEventIds = mergeEventIds(existing.supporting_event_ids, fact.supporting_event_ids);
updateRel.run(
finalCount,
confidenceFromCount(finalCount),
finalLast,
finalEventIds,
fact.company_id,
relationshipType,
toEntity
);
} else {
finalCount = fact.confirmation_count || 1;
finalFirst = fact.first_seen_at || now;
finalLast = fact.last_seen_at || now;
finalEventIds = fact.supporting_event_ids || "[]";
insertRel.run(
fact.company_id,
relationshipType,
toEntity,
toCompanyId,
confidenceFromCount(finalCount),
finalCount,
finalFirst,
finalLast,
finalEventIds
);
}
upserted++;
// reciprocal edge — only where both companies are tracked
if (toCompanyId) {
const fromCompany = getCompanyById.get(fact.company_id);
if (!fromCompany) continue;
const recipType = RECIPROCAL[relationshipType] || "partner";
const recipToEntity = fromCompany.name;
const recipExists = getExistingReciprocal.get(toCompanyId, recipType, recipToEntity);
if (!recipExists) {
insertReciprocal.run(
toCompanyId,
recipType,
recipToEntity,
fact.company_id,
confidenceFromCount(finalCount),
finalCount,
finalFirst,
finalLast,
finalEventIds
);
reciprocals++;
}
}
}
});
processAll();
if (upserted > 0 || reciprocals > 0) {
console.log(`[graph] cycle complete — ${upserted} edges upserted, ${reciprocals} reciprocals added`);
} else {
console.log("[graph] cycle complete — no new data");
}
} catch (err) {
console.error("[graph] cycle error:", err.message);
}
await sleep(loopDelay);
}
}
function sleep(ms) {
return new Promise(r => setTimeout(r, ms));
}
module.exports = { runGraphWorker };

View file

@ -6,6 +6,7 @@ const { runQueueFeeder } = require("./queueFeeder");
const { runAugorWorker } = require("./augorWorker"); const { runAugorWorker } = require("./augorWorker");
const { ensureCompanyEmbeddings } = require("./embeddings"); const { ensureCompanyEmbeddings } = require("./embeddings");
const { runConsolidationWorker } = require("./consolidationWorker"); const { runConsolidationWorker } = require("./consolidationWorker");
const { runGraphWorker } = require("./graphWorker");
const configPath = path.resolve(__dirname, "../config.json"); const configPath = path.resolve(__dirname, "../config.json");
@ -58,6 +59,11 @@ runConsolidationWorker(archiveDb, intelligenceDb, config).catch(err => {
process.exit(1); process.exit(1);
}); });
runGraphWorker(archiveDb, intelligenceDb, config).catch(err => {
console.error("[graph] fatal:", err);
process.exit(1);
});
process.on("SIGINT", () => { process.on("SIGINT", () => {
console.log("[intelligence] shutting down"); console.log("[intelligence] shutting down");
process.exit(0); process.exit(0);

View file

@ -307,6 +307,57 @@ async function adminRoutes(fastify) {
return { total, rows }; return { total, rows };
}); });
// intelligence graph — nodes + edges from company_relationships
fastify.get('/admin/api/intelligence/graph', async (request, reply) => {
if (!checkAuth(request, reply)) return;
const idb = getIntelligenceDb();
if (!idb) return { nodes: [], edges: [] };
const edges = idb.prepare(`SELECT * FROM company_relationships`).all();
const trackedIds = new Set();
for (const e of edges) {
trackedIds.add(e.from_company_id);
if (e.to_company_id) trackedIds.add(e.to_company_id);
}
const allTracked = idb.prepare(`SELECT id, name, ticker FROM tracked_companies`).all();
// untracked entities — appear only as to_entity with no to_company_id
const untrackedSeen = new Set();
for (const e of edges) {
if (!e.to_company_id && e.to_entity) untrackedSeen.add(e.to_entity);
}
const nodes = [
...allTracked
.filter(c => trackedIds.has(c.id))
.map(c => ({ id: c.id, name: c.name, ticker: c.ticker, tracked: true })),
...[...untrackedSeen].map(name => ({ id: null, name, ticker: null, tracked: false })),
];
return { nodes, edges };
});
// per-company facts for the graph sidebar
fastify.get('/admin/api/intelligence/facts/:company_id', async (request, reply) => {
if (!checkAuth(request, reply)) return;
const idb = getIntelligenceDb();
if (!idb) return [];
const companyId = parseInt(request.params.company_id, 10);
return idb.prepare(`
SELECT type, claim, confidence, confirmation_count
FROM company_facts
WHERE company_id = ?
ORDER BY confirmation_count DESC
LIMIT 10
`).all(companyId);
});
// raw sql console // raw sql console
fastify.post('/admin/api/sql', async (request, reply) => { fastify.post('/admin/api/sql', async (request, reply) => {
if (!checkAuth(request, reply)) return; if (!checkAuth(request, reply)) return;