use std::fs; use std::path::{Path, PathBuf}; use chrono::NaiveDate; use directories::ProjectDirs; use rusqlite::{Connection, params}; use crate::config::get_config_value; use crate::error::IdxError; use crate::ownership::types::{ BingHolding, ChangeRow, ChangeType, ConcentrationMetrics, CrossHolderRow, Entity, EntityHoldings, EntityTickerRow, FlowSignal, HolderRow, InstitutionalFlow, KseiHolding, Locality, OwnershipRelease, OwnershipSource, Ticker, TickerOwnership, UnresolvedRow, }; /// Ownership schema version 1 DDL. pub const SCHEMA_V1: &str = r#" PRAGMA foreign_keys = ON; CREATE TABLE IF NOT EXISTS ownership_releases ( id INTEGER PRIMARY KEY, source_url TEXT, sha256 TEXT NOT NULL UNIQUE, as_of_date TEXT NOT NULL, row_count INTEGER NOT NULL, imported_at INTEGER NOT NULL ); CREATE TABLE IF NOT EXISTS entities ( id INTEGER PRIMARY KEY, canonical_name TEXT NOT NULL, entity_type TEXT, country TEXT, created_at INTEGER NOT NULL, updated_at INTEGER NOT NULL ); CREATE TABLE IF NOT EXISTS entity_aliases ( id INTEGER PRIMARY KEY, entity_id INTEGER NOT NULL REFERENCES entities(id), raw_name TEXT NOT NULL, source TEXT NOT NULL, confidence REAL NOT NULL DEFAULT 1.0, method TEXT NOT NULL, UNIQUE(raw_name, source) ); CREATE INDEX IF NOT EXISTS idx_aliases_entity ON entity_aliases(entity_id); CREATE INDEX IF NOT EXISTS idx_aliases_name ON entity_aliases(raw_name); CREATE TABLE IF NOT EXISTS tickers ( id INTEGER PRIMARY KEY, code TEXT NOT NULL UNIQUE, name TEXT, sector TEXT ); CREATE TABLE IF NOT EXISTS ksei_holdings ( id INTEGER PRIMARY KEY, ticker_id INTEGER NOT NULL REFERENCES tickers(id), entity_id INTEGER REFERENCES entities(id), raw_investor_name TEXT NOT NULL, investor_type TEXT, locality TEXT, nationality TEXT, domicile TEXT, holdings_scripless INTEGER NOT NULL, holdings_scrip INTEGER NOT NULL, total_shares INTEGER NOT NULL, percentage_bps INTEGER NOT NULL, report_date TEXT NOT NULL, release_sha256 TEXT NOT NULL, UNIQUE(release_sha256, ticker_id, raw_investor_name) ); CREATE INDEX IF NOT EXISTS idx_ksei_ticker ON ksei_holdings(ticker_id); CREATE INDEX IF NOT EXISTS idx_ksei_entity ON ksei_holdings(entity_id); CREATE INDEX IF NOT EXISTS idx_ksei_date ON ksei_holdings(report_date); CREATE INDEX IF NOT EXISTS idx_ksei_pct ON ksei_holdings(percentage_bps DESC); CREATE TABLE IF NOT EXISTS bing_holdings ( id INTEGER PRIMARY KEY, ticker_id INTEGER NOT NULL REFERENCES tickers(id), entity_id INTEGER REFERENCES entities(id), raw_investor_name TEXT NOT NULL, investor_type TEXT, shares_held INTEGER, shares_changed INTEGER, pct_ownership_bps INTEGER, value_usd INTEGER, report_date TEXT NOT NULL, signal TEXT NOT NULL, fetched_at INTEGER NOT NULL, UNIQUE(ticker_id, raw_investor_name, report_date, signal) ); CREATE INDEX IF NOT EXISTS idx_bing_ticker ON bing_holdings(ticker_id); CREATE INDEX IF NOT EXISTS idx_bing_entity ON bing_holdings(entity_id); CREATE INDEX IF NOT EXISTS idx_bing_date ON bing_holdings(report_date); "#; /// Ensure ownership schema is migrated and ready for use. pub fn ensure_schema(conn: &Connection) -> Result<(), IdxError> { let user_version: i64 = conn .query_row("PRAGMA user_version", [], |row| row.get(0)) .map_err(|e| IdxError::DatabaseError(e.to_string()))?; if conn.path().is_some() { let _ = conn.pragma_update(None, "journal_mode", "WAL"); } conn.pragma_update(None, "foreign_keys", "ON") .map_err(|e| IdxError::DatabaseError(e.to_string()))?; if user_version < 1 { conn.execute_batch(SCHEMA_V1) .map_err(|e| IdxError::DatabaseError(e.to_string()))?; conn.pragma_update(None, "user_version", 1) .map_err(|e| IdxError::DatabaseError(e.to_string()))?; } Ok(()) } /// Open ownership database connection and run idempotent schema migration. pub fn open_db() -> Result { open_db_at(&db_path()?) } /// Open the ownership database at `db_path`, creating it and migrating the schema. /// Where the database file really lives: symlinks are followed by reading /// the link, so this works even while the target does not exist (a fresh /// install, or mid-replacement during sync). Creates the parent directory. pub(crate) fn resolve_db_path(path: &Path) -> Result { let mut current = path.to_path_buf(); for _ in 0..16 { let parent = match current.parent() { Some(parent) if !parent.as_os_str().is_empty() => parent.to_path_buf(), _ => PathBuf::from("."), }; fs::create_dir_all(&parent).map_err(|e| IdxError::Io(e.to_string()))?; let parent = fs::canonicalize(&parent).map_err(|e| IdxError::Io(e.to_string()))?; let name = current.file_name().ok_or_else(|| { IdxError::ConfigError(format!( "ownership database path has no file name: {}", path.display() )) })?; let candidate = parent.join(name); match fs::symlink_metadata(&candidate) { Ok(meta) if meta.file_type().is_symlink() => { let target = fs::read_link(&candidate).map_err(|e| IdxError::Io(e.to_string()))?; current = if target.is_absolute() { target } else { parent.join(target) }; } _ => return Ok(candidate), } } Err(IdxError::Io(format!( "too many symlink levels resolving {}", path.display() ))) } pub fn open_db_at(db_path: &Path) -> Result { if let Some(parent) = db_path.parent() { fs::create_dir_all(parent).map_err(|e| IdxError::DatabaseError(e.to_string()))?; } let conn = Connection::open(db_path).map_err(|e| IdxError::DatabaseError(e.to_string()))?; ensure_schema(&conn)?; Ok(conn) } /// Insert or get a ticker by code. Returns ticker_id. pub fn upsert_ticker(conn: &Connection, code: &str, name: Option<&str>) -> Result { conn.execute( "INSERT INTO tickers (code, name) VALUES (?1, ?2) ON CONFLICT(code) DO UPDATE SET name = COALESCE(excluded.name, tickers.name)", params![code, name], ) .map_err(|e| IdxError::DatabaseError(e.to_string()))?; conn.query_row( "SELECT id FROM tickers WHERE code = ?1", params![code], |row| row.get(0), ) .map_err(|e| IdxError::DatabaseError(e.to_string())) } fn insert_ksei_holdings_rows( conn: &Connection, holdings: &[KseiHolding], ) -> Result { if holdings.is_empty() { return Ok(0); } let mut inserted = 0usize; for h in holdings { let locality = h.locality.map(locality_to_db); let investor_type = h.investor_type.as_ref().map(|v| v.0.as_str()); let changed = conn .execute( "INSERT OR IGNORE INTO ksei_holdings ( ticker_id, entity_id, raw_investor_name, investor_type, locality, nationality, domicile, holdings_scripless, holdings_scrip, total_shares, percentage_bps, report_date, release_sha256 ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13)", params![ h.ticker_id, h.entity_id, h.raw_investor_name, investor_type, locality, h.nationality, h.domicile, h.holdings_scripless, h.holdings_scrip, h.total_shares, h.percentage_bps, h.report_date.format("%Y-%m-%d").to_string(), h.release_sha256, ], ) .map_err(|e| IdxError::DatabaseError(e.to_string())); inserted += changed?; } Ok(inserted) } /// Bulk insert KSEI holdings within a transaction. /// Uses INSERT OR IGNORE for dedup (unique on release_sha256 + ticker_id + raw_investor_name). pub fn insert_ksei_holdings( conn: &Connection, holdings: &[KseiHolding], ) -> Result { if holdings.is_empty() { return Ok(0); } conn.execute("BEGIN IMMEDIATE", []) .map_err(|e| IdxError::DatabaseError(e.to_string()))?; let inserted = match insert_ksei_holdings_rows(conn, holdings) { Ok(inserted) => inserted, Err(err) => { let _ = conn.execute("ROLLBACK", []); return Err(err); } }; conn.execute("COMMIT", []) .map_err(|e| IdxError::DatabaseError(e.to_string()))?; Ok(inserted) } /// Bulk insert Bing holdings within a transaction. pub fn insert_bing_holdings( conn: &Connection, holdings: &[BingHolding], ) -> Result { if holdings.is_empty() { return Ok(0); } conn.execute("BEGIN IMMEDIATE", []) .map_err(|e| IdxError::DatabaseError(e.to_string()))?; let mut inserted = 0usize; for h in holdings { let changed = conn .execute( "INSERT OR IGNORE INTO bing_holdings ( ticker_id, entity_id, raw_investor_name, investor_type, shares_held, shares_changed, pct_ownership_bps, value_usd, report_date, signal, fetched_at ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11)", params![ h.ticker_id, h.entity_id, h.raw_investor_name, h.investor_type, h.shares_held, h.shares_changed, h.pct_ownership_bps, h.value_usd, h.report_date.format("%Y-%m-%d").to_string(), flow_signal_to_db(h.signal), h.fetched_at, ], ) .map_err(|e| IdxError::DatabaseError(e.to_string())); match changed { Ok(n) => inserted += n, Err(err) => { let _ = conn.execute("ROLLBACK", []); return Err(err); } } } conn.execute("COMMIT", []) .map_err(|e| IdxError::DatabaseError(e.to_string()))?; Ok(inserted) } /// Record a KSEI release import. Returns release_id. pub fn insert_release(conn: &Connection, release: &OwnershipRelease) -> Result { conn.execute( "INSERT INTO ownership_releases (source_url, sha256, as_of_date, row_count, imported_at) VALUES (?1, ?2, ?3, ?4, ?5)", params![ release.source_url, release.sha256, release.as_of_date.format("%Y-%m-%d").to_string(), i64::try_from(release.row_count) .map_err(|e| IdxError::DatabaseError(format!("invalid row_count: {e}")))?, release.imported_at, ], ) .map_err(|e| IdxError::DatabaseError(e.to_string()))?; Ok(conn.last_insert_rowid()) } /// Insert or replace a complete KSEI release and its rows atomically. pub fn write_ksei_release( conn: &Connection, release: &OwnershipRelease, holdings: &[KseiHolding], replace_existing: bool, ) -> Result { conn.execute("BEGIN IMMEDIATE", []) .map_err(|e| IdxError::DatabaseError(e.to_string()))?; let result = (|| { if replace_existing { conn.execute( "DELETE FROM ksei_holdings WHERE release_sha256 = ?1", params![&release.sha256], ) .map_err(|e| IdxError::DatabaseError(e.to_string()))?; conn.execute( "DELETE FROM ownership_releases WHERE sha256 = ?1", params![&release.sha256], ) .map_err(|e| IdxError::DatabaseError(e.to_string()))?; } let inserted = insert_ksei_holdings_rows(conn, holdings)?; // Record the rows stored, not parsed: INSERT OR IGNORE drops repeated // ticker/investor rows, and snapshot validation counts what is stored. let stored = OwnershipRelease { row_count: inserted, ..release.clone() }; let _ = insert_release(conn, &stored)?; Ok(inserted) })(); match result { Ok(inserted) => { conn.execute("COMMIT", []) .map_err(|e| IdxError::DatabaseError(e.to_string()))?; Ok(inserted) } Err(err) => { let _ = conn.execute("ROLLBACK", []); Err(err) } } } /// Check if a release with this SHA-256 already exists. pub fn release_exists(conn: &Connection, sha256: &str) -> Result { let exists: i64 = conn .query_row( "SELECT EXISTS(SELECT 1 FROM ownership_releases WHERE sha256 = ?1)", params![sha256], |row| row.get(0), ) .map_err(|e| IdxError::DatabaseError(e.to_string()))?; Ok(exists == 1) } /// Get combined KSEI + Bing holdings for a ticker. /// Returns TickerOwnership with merged holders sorted by percentage desc, /// concentration metrics, and Bing flow data. pub fn query_ticker_holdings(conn: &Connection, code: &str) -> Result { let ticker = query_ticker(conn, code)?.ok_or_else(|| IdxError::SymbolNotFound(code.to_string()))?; let latest_ksei = latest_ksei_release(conn)?; let ksei_as_of = latest_ksei.as_ref().map(|release| release.as_of_date); let bing_as_of = conn .query_row( "SELECT MAX(report_date) FROM bing_holdings WHERE ticker_id = ?1", params![ticker.id], |row| row.get::<_, Option>(0), ) .map_err(|e| IdxError::DatabaseError(e.to_string()))?; let mut holders: Vec = Vec::new(); if let Some(latest_ksei) = latest_ksei.as_ref() { let mut stmt = conn .prepare( "SELECT k.entity_id, COALESCE(e.canonical_name, k.raw_investor_name), k.investor_type, k.locality, k.total_shares, k.percentage_bps FROM ksei_holdings k LEFT JOIN entities e ON e.id = k.entity_id WHERE k.ticker_id = ?1 AND k.release_sha256 = ?2 ORDER BY k.percentage_bps DESC, k.total_shares DESC", ) .map_err(|e| IdxError::DatabaseError(e.to_string()))?; let rows = stmt .query_map(params![ticker.id, &latest_ksei.sha256], |row| { Ok(( row.get::<_, Option>(0)?, row.get::<_, String>(1)?, row.get::<_, Option>(2)?, row.get::<_, Option>(3)?, row.get::<_, i64>(4)?, row.get::<_, i64>(5)?, )) }) .map_err(|e| IdxError::DatabaseError(e.to_string()))?; for row in rows { let (entity_id, name, investor_type, locality_raw, shares, percentage_bps) = row.map_err(|e| IdxError::DatabaseError(e.to_string()))?; holders.push(HolderRow { rank: 0, source: OwnershipSource::Ksei, name, entity_id, investor_type, locality: locality_raw.as_deref().and_then(locality_from_db), shares, percentage_bps, signal: None, }); } } if let Some(bing_as_of) = bing_as_of.as_ref() { let mut stmt = conn .prepare( "SELECT b.entity_id, COALESCE(e.canonical_name, b.raw_investor_name), b.investor_type, COALESCE(b.shares_held, 0), COALESCE(b.pct_ownership_bps, 0), b.signal FROM bing_holdings b LEFT JOIN entities e ON e.id = b.entity_id WHERE b.ticker_id = ?1 AND b.report_date = ?2 AND b.signal = 'holder' ORDER BY COALESCE(b.pct_ownership_bps, 0) DESC, COALESCE(b.shares_held, 0) DESC", ) .map_err(|e| IdxError::DatabaseError(e.to_string()))?; let rows = stmt .query_map(params![ticker.id, bing_as_of], |row| { Ok(( row.get::<_, Option>(0)?, row.get::<_, String>(1)?, row.get::<_, Option>(2)?, row.get::<_, i64>(3)?, row.get::<_, i64>(4)?, row.get::<_, String>(5)?, )) }) .map_err(|e| IdxError::DatabaseError(e.to_string()))?; for row in rows { let (entity_id, name, investor_type, shares, percentage_bps, signal_raw) = row.map_err(|e| IdxError::DatabaseError(e.to_string()))?; holders.push(HolderRow { rank: 0, source: OwnershipSource::Bing, name, entity_id, investor_type, locality: None, shares, percentage_bps, signal: flow_signal_from_db(&signal_raw), }); } } holders.sort_by(|a, b| { b.percentage_bps .cmp(&a.percentage_bps) .then_with(|| b.shares.cmp(&a.shares)) }); for (idx, holder) in holders.iter_mut().enumerate() { holder.rank = idx + 1; } let percentages = holders.iter().map(|h| h.percentage_bps).collect::>(); let concentration = compute_concentration(&percentages); let flow = query_bing_flow(conn, ticker.id)?; Ok(TickerOwnership { ticker, ksei_as_of, bing_as_of, holders, concentration, flow, }) } /// Get all holdings for an entity across tickers. pub fn query_entity_holdings( conn: &Connection, entity_id: i64, ) -> Result { let entity = query_entity(conn, entity_id)? .ok_or_else(|| IdxError::SymbolNotFound(format!("entity:{entity_id}")))?; let mut holdings = Vec::new(); if let Some(latest_ksei) = latest_ksei_release(conn)? { let mut stmt = conn .prepare( "SELECT t.id, t.code, t.name, t.sector, k.total_shares, k.percentage_bps, k.report_date FROM ksei_holdings k JOIN tickers t ON t.id = k.ticker_id WHERE k.entity_id = ?1 AND k.release_sha256 = ?2", ) .map_err(|e| IdxError::DatabaseError(e.to_string()))?; let rows = stmt .query_map(params![entity_id, latest_ksei.sha256], |row| { Ok(EntityTickerRow { ticker: Ticker { id: row.get(0)?, code: row.get(1)?, name: row.get(2)?, sector: row.get(3)?, }, source: OwnershipSource::Ksei, shares: row.get(4)?, percentage_bps: row.get(5)?, report_date: parse_iso_date(&row.get::<_, String>(6)?) .map_err(|e| rusqlite::Error::ToSqlConversionFailure(Box::new(e)))?, }) }) .map_err(|e| IdxError::DatabaseError(e.to_string()))?; for row in rows { holdings.push(row.map_err(|e| IdxError::DatabaseError(e.to_string()))?); } } { let mut stmt = conn .prepare( "WITH latest_bing AS ( SELECT ticker_id, MAX(report_date) AS report_date FROM bing_holdings GROUP BY ticker_id ) SELECT t.id, t.code, t.name, t.sector, COALESCE(b.shares_held, 0), COALESCE(b.pct_ownership_bps, 0), b.report_date FROM bing_holdings b JOIN latest_bing lb ON lb.ticker_id = b.ticker_id AND lb.report_date = b.report_date JOIN tickers t ON t.id = b.ticker_id WHERE b.entity_id = ?1 AND b.signal = 'holder'", ) .map_err(|e| IdxError::DatabaseError(e.to_string()))?; let rows = stmt .query_map(params![entity_id], |row| { Ok(EntityTickerRow { ticker: Ticker { id: row.get(0)?, code: row.get(1)?, name: row.get(2)?, sector: row.get(3)?, }, source: OwnershipSource::Bing, shares: row.get(4)?, percentage_bps: row.get(5)?, report_date: parse_iso_date(&row.get::<_, String>(6)?) .map_err(|e| rusqlite::Error::ToSqlConversionFailure(Box::new(e)))?, }) }) .map_err(|e| IdxError::DatabaseError(e.to_string()))?; for row in rows { holdings.push(row.map_err(|e| IdxError::DatabaseError(e.to_string()))?); } } holdings.sort_by(|a, b| { b.percentage_bps .cmp(&a.percentage_bps) .then_with(|| a.ticker.code.cmp(&b.ticker.code)) }); let ticker_count = holdings .iter() .map(|h| h.ticker.id) .collect::>() .len(); Ok(EntityHoldings { entity, ticker_count, holdings, }) } /// Rank entities by number of tickers they hold (cross-ownership breadth). pub fn query_cross_holders( conn: &Connection, min_tickers: usize, limit: usize, ) -> Result, IdxError> { let mut aggregates: std::collections::BTreeMap< i64, (Entity, std::collections::BTreeSet, i64), > = std::collections::BTreeMap::new(); if let Some(latest_ksei) = latest_ksei_release(conn)? { let mut stmt = conn .prepare( "SELECT e.id, e.canonical_name, e.entity_type, e.country, k.ticker_id, k.percentage_bps FROM ksei_holdings k JOIN entities e ON e.id = k.entity_id WHERE k.entity_id IS NOT NULL AND k.release_sha256 = ?1", ) .map_err(|e| IdxError::DatabaseError(e.to_string()))?; let rows = stmt .query_map(params![latest_ksei.sha256], |row| { Ok(( Entity { id: row.get(0)?, canonical_name: row.get(1)?, entity_type: row.get(2)?, country: row.get(3)?, }, row.get::<_, i64>(4)?, row.get::<_, i64>(5)?, )) }) .map_err(|e| IdxError::DatabaseError(e.to_string()))?; for row in rows { let (entity, ticker_id, percentage_bps) = row.map_err(|e| IdxError::DatabaseError(e.to_string()))?; let entry = aggregates .entry(entity.id) .or_insert_with(|| (entity, std::collections::BTreeSet::new(), 0)); entry.1.insert(ticker_id); entry.2 += percentage_bps; } } { let mut stmt = conn .prepare( "WITH latest_bing AS ( SELECT ticker_id, MAX(report_date) AS report_date FROM bing_holdings GROUP BY ticker_id ) SELECT e.id, e.canonical_name, e.entity_type, e.country, b.ticker_id, COALESCE(b.pct_ownership_bps, 0) FROM bing_holdings b JOIN latest_bing lb ON lb.ticker_id = b.ticker_id AND lb.report_date = b.report_date JOIN entities e ON e.id = b.entity_id WHERE b.entity_id IS NOT NULL AND b.signal = 'holder'", ) .map_err(|e| IdxError::DatabaseError(e.to_string()))?; let rows = stmt .query_map([], |row| { Ok(( Entity { id: row.get(0)?, canonical_name: row.get(1)?, entity_type: row.get(2)?, country: row.get(3)?, }, row.get::<_, i64>(4)?, row.get::<_, i64>(5)?, )) }) .map_err(|e| IdxError::DatabaseError(e.to_string()))?; for row in rows { let (entity, ticker_id, percentage_bps) = row.map_err(|e| IdxError::DatabaseError(e.to_string()))?; let entry = aggregates .entry(entity.id) .or_insert_with(|| (entity, std::collections::BTreeSet::new(), 0)); entry.1.insert(ticker_id); entry.2 += percentage_bps; } } let mut out = aggregates .into_values() .filter_map(|(entity, tickers, total_bps)| { (tickers.len() >= min_tickers).then_some(CrossHolderRow { entity, ticker_count: tickers.len(), total_bps, }) }) .collect::>(); out.sort_by(|a, b| { b.ticker_count .cmp(&a.ticker_count) .then_with(|| b.total_bps.cmp(&a.total_bps)) .then_with(|| a.entity.canonical_name.cmp(&b.entity.canonical_name)) }); out.truncate(limit); Ok(out) } /// Rank tickers by ownership concentration. /// sort_by: "top1", "top3", "hhi" pub fn query_concentration( conn: &Connection, sort_by: &str, limit: usize, ) -> Result, IdxError> { let valid = ["top1", "top3", "hhi"]; if !valid.contains(&sort_by) { return Err(IdxError::ParseError(format!( "invalid sort_by '{sort_by}', expected one of: top1, top3, hhi" ))); } let Some(latest_ksei) = latest_ksei_release(conn)? else { return Ok(Vec::new()); }; let mut stmt = conn .prepare( "SELECT t.code, k.percentage_bps FROM tickers t JOIN ksei_holdings k ON k.ticker_id = t.id WHERE k.release_sha256 = ?1 ORDER BY t.code ASC, k.percentage_bps DESC", ) .map_err(|e| IdxError::DatabaseError(e.to_string()))?; let rows = stmt .query_map(params![latest_ksei.sha256], |row| { Ok((row.get::<_, String>(0)?, row.get::<_, i64>(1)?)) }) .map_err(|e| IdxError::DatabaseError(e.to_string()))?; let mut grouped: std::collections::BTreeMap> = std::collections::BTreeMap::new(); for row in rows { let (code, bps) = row.map_err(|e| IdxError::DatabaseError(e.to_string()))?; grouped.entry(code).or_default().push(bps); } let mut ranked: Vec<(String, ConcentrationMetrics)> = grouped .into_iter() .map(|(code, bps)| (code, compute_concentration(&bps))) .collect(); ranked.sort_by(|a, b| { let am = &a.1; let bm = &b.1; match sort_by { "top1" => bm.top1_bps.cmp(&am.top1_bps), "top3" => bm.top3_bps.cmp(&am.top3_bps), "hhi" => bm.hhi.cmp(&am.hhi), _ => std::cmp::Ordering::Equal, } .then_with(|| a.0.cmp(&b.0)) }); ranked.truncate(limit); Ok(ranked) } /// List all imported releases. pub fn query_releases(conn: &Connection) -> Result, IdxError> { let mut stmt = conn .prepare( "SELECT id, source_url, sha256, as_of_date, row_count, imported_at FROM ownership_releases ORDER BY as_of_date DESC, imported_at DESC, id DESC", ) .map_err(|e| IdxError::DatabaseError(e.to_string()))?; let rows = stmt .query_map([], |row| { Ok(OwnershipRelease { id: row.get(0)?, source_url: row.get(1)?, sha256: row.get(2)?, as_of_date: parse_iso_date(&row.get::<_, String>(3)?) .map_err(|e| rusqlite::Error::ToSqlConversionFailure(Box::new(e)))?, row_count: row.get::<_, i64>(4)? as usize, imported_at: row.get(5)?, }) }) .map_err(|e| IdxError::DatabaseError(e.to_string()))?; let mut out = Vec::new(); for row in rows { out.push(row.map_err(|e| IdxError::DatabaseError(e.to_string()))?); } Ok(out) } /// Get Bing institutional flow for a ticker. pub fn query_bing_flow( conn: &Connection, ticker_id: i64, ) -> Result, IdxError> { let latest = conn .query_row( "SELECT MAX(report_date) FROM bing_holdings WHERE ticker_id = ?1", params![ticker_id], |row| row.get::<_, Option>(0), ) .map_err(|e| IdxError::DatabaseError(e.to_string()))?; let Some(period) = latest else { return Ok(None); }; let mut stmt = conn .prepare( "SELECT b.entity_id, COALESCE(e.canonical_name, b.raw_investor_name), b.investor_type, COALESCE(b.shares_held, 0), COALESCE(b.pct_ownership_bps, 0), b.signal FROM bing_holdings b LEFT JOIN entities e ON e.id = b.entity_id WHERE b.ticker_id = ?1 AND b.report_date = ?2 ORDER BY COALESCE(b.pct_ownership_bps, 0) DESC, COALESCE(b.shares_held, 0) DESC", ) .map_err(|e| IdxError::DatabaseError(e.to_string()))?; let rows = stmt .query_map(params![ticker_id, &period], |row| { Ok(( row.get::<_, Option>(0)?, row.get::<_, String>(1)?, row.get::<_, Option>(2)?, row.get::<_, i64>(3)?, row.get::<_, i64>(4)?, row.get::<_, String>(5)?, )) }) .map_err(|e| IdxError::DatabaseError(e.to_string()))?; let mut top_buyers = Vec::new(); let mut top_sellers = Vec::new(); let mut new_positions = Vec::new(); let mut exited = Vec::new(); let mut rank_counter = 1usize; for row in rows { let (entity_id, name, investor_type, shares, percentage_bps, signal_raw) = row.map_err(|e| IdxError::DatabaseError(e.to_string()))?; let Some(signal) = flow_signal_from_db(&signal_raw) else { continue; }; let holder = HolderRow { rank: rank_counter, source: OwnershipSource::Bing, name, entity_id, investor_type, locality: None, shares, percentage_bps, signal: Some(signal), }; rank_counter += 1; match signal { FlowSignal::Buyer => top_buyers.push(holder), FlowSignal::Seller => top_sellers.push(holder), FlowSignal::NewPosition => new_positions.push(holder), FlowSignal::Exited => exited.push(holder), FlowSignal::Holder => {} } } Ok(Some(InstitutionalFlow { period, top_buyers, top_sellers, new_positions, exited, })) } /// The release used for a report date: the latest imported one, as /// `latest_ksei_release` picks, so a corrected file supersedes the original. /// The latest import holding rows reported on `date`. A file can carry rows /// from several report dates, so a release is matched by its rows rather /// than by its `as_of_date` (the latest date in it). fn release_for_date(conn: &Connection, date: &str) -> Result, IdxError> { conn.query_row( "SELECT r.sha256 FROM ownership_releases r WHERE EXISTS ( SELECT 1 FROM ksei_holdings k WHERE k.release_sha256 = r.sha256 AND k.report_date = ?1 ) ORDER BY r.imported_at DESC, r.id DESC LIMIT 1", params![date], |row| row.get::<_, String>(0), ) .map(Some) .or_else(|e| match e { rusqlite::Error::QueryReturnedNoRows => Ok(None), other => Err(IdxError::DatabaseError(other.to_string())), }) } /// Compare two KSEI releases by date and return ownership changes. /// Finds: new holders, exited holders, percentage increases/decreases. /// Each date reads its rows from one release, and a holder's accounts in a ticker are /// summed, so an investor with two accounts compares by its total stake. pub fn query_changes( conn: &Connection, from_date: &str, to_date: &str, ) -> Result, IdxError> { let release = |date: &str| { release_for_date(conn, date)?.ok_or_else(|| { IdxError::InvalidInput(format!( "no KSEI release imported for {date}; list the imported dates with `idx ownership releases` or the ownership_status tool" )) }) }; let (from_release, to_release) = (release(from_date)?, release(to_date)?); let mut stmt = conn .prepare( "WITH from_snapshot AS ( SELECT t.code AS ticker_code, MIN(COALESCE(e.canonical_name, k.raw_investor_name)) AS entity_name, CASE WHEN k.entity_id IS NOT NULL THEN 'id:' || k.entity_id ELSE 'raw:' || UPPER(TRIM(k.raw_investor_name)) END AS holder_key, SUM(k.percentage_bps) AS old_bps FROM ksei_holdings k JOIN tickers t ON t.id = k.ticker_id LEFT JOIN entities e ON e.id = k.entity_id WHERE k.release_sha256 = ?1 AND k.report_date = ?3 GROUP BY t.code, holder_key ), to_snapshot AS ( SELECT t.code AS ticker_code, MIN(COALESCE(e.canonical_name, k.raw_investor_name)) AS entity_name, CASE WHEN k.entity_id IS NOT NULL THEN 'id:' || k.entity_id ELSE 'raw:' || UPPER(TRIM(k.raw_investor_name)) END AS holder_key, SUM(k.percentage_bps) AS new_bps FROM ksei_holdings k JOIN tickers t ON t.id = k.ticker_id LEFT JOIN entities e ON e.id = k.entity_id WHERE k.release_sha256 = ?2 AND k.report_date = ?4 GROUP BY t.code, holder_key ), keys AS ( SELECT ticker_code, holder_key FROM from_snapshot UNION SELECT ticker_code, holder_key FROM to_snapshot ) SELECT k.ticker_code, COALESCE(ts.entity_name, fs.entity_name) AS entity_name, fs.old_bps, ts.new_bps FROM keys k LEFT JOIN from_snapshot fs ON fs.ticker_code = k.ticker_code AND fs.holder_key = k.holder_key LEFT JOIN to_snapshot ts ON ts.ticker_code = k.ticker_code AND ts.holder_key = k.holder_key WHERE (fs.old_bps IS NULL AND ts.new_bps IS NOT NULL) OR (fs.old_bps IS NOT NULL AND ts.new_bps IS NULL) OR (fs.old_bps IS NOT NULL AND ts.new_bps IS NOT NULL AND fs.old_bps <> ts.new_bps) ORDER BY k.ticker_code ASC, ABS(COALESCE(ts.new_bps, 0) - COALESCE(fs.old_bps, 0)) DESC, entity_name ASC", ) .map_err(|e| IdxError::DatabaseError(e.to_string()))?; let rows = stmt .query_map( params![from_release, to_release, from_date, to_date], |row| { Ok(( row.get::<_, String>(0)?, row.get::<_, String>(1)?, row.get::<_, Option>(2)?, row.get::<_, Option>(3)?, )) }, ) .map_err(|e| IdxError::DatabaseError(e.to_string()))?; let mut out = Vec::new(); for row in rows { let (ticker_code, entity_name, old_bps, new_bps) = row.map_err(|e| IdxError::DatabaseError(e.to_string()))?; let change_type = match (old_bps, new_bps) { (None, Some(_)) => ChangeType::New, (Some(_), None) => ChangeType::Exited, (Some(old), Some(new)) if new > old => ChangeType::Increased, (Some(_), Some(_)) => ChangeType::Decreased, (None, None) => continue, }; let delta_bps = new_bps.unwrap_or(0) - old_bps.unwrap_or(0); out.push(ChangeRow { ticker_code, entity_name, change_type, old_bps, new_bps, delta_bps, }); } Ok(out) } /// List aliases that are unresolved in holdings or have low-confidence mappings. pub fn list_unresolved(conn: &Connection, limit: usize) -> Result, IdxError> { let limit_i64 = i64::try_from(limit).map_err(|e| IdxError::DatabaseError(format!("invalid limit: {e}")))?; let mut stmt = conn .prepare( "SELECT DISTINCT h.raw_name, h.source, t.code, e.canonical_name, a.confidence FROM ( SELECT raw_investor_name AS raw_name, 'ksei' AS source, ticker_id, entity_id FROM ksei_holdings UNION ALL SELECT raw_investor_name AS raw_name, 'bing' AS source, ticker_id, entity_id FROM bing_holdings ) h JOIN tickers t ON t.id = h.ticker_id LEFT JOIN entity_aliases a ON a.raw_name = h.raw_name AND a.source = h.source LEFT JOIN entities e ON e.id = COALESCE(a.entity_id, h.entity_id) WHERE h.entity_id IS NULL OR a.entity_id IS NULL OR COALESCE(a.confidence, 0.0) < 0.8 ORDER BY COALESCE(a.confidence, 0.0) ASC, h.raw_name ASC LIMIT ?1", ) .map_err(|e| IdxError::DatabaseError(e.to_string()))?; let rows = stmt .query_map(params![limit_i64], |row| { Ok(UnresolvedRow { raw_name: row.get(0)?, source: row.get(1)?, ticker_code: row.get(2)?, current_entity: row.get(3)?, confidence: row.get(4)?, }) }) .map_err(|e| IdxError::DatabaseError(e.to_string()))?; let mut out = Vec::new(); for row in rows { out.push(row.map_err(|e| IdxError::DatabaseError(e.to_string()))?); } Ok(out) } /// Manually map a raw investor name to a canonical entity. /// Creates the entity if it does not exist, then creates/updates aliases and unresolved holdings. pub fn manual_map(conn: &Connection, raw_name: &str, canonical_name: &str) -> Result<(), IdxError> { let raw_name = raw_name.trim(); let canonical_name = canonical_name.trim(); if raw_name.is_empty() || canonical_name.is_empty() { return Err(IdxError::ParseError( "alias and canonical entity must be non-empty".to_string(), )); } conn.execute("BEGIN IMMEDIATE", []) .map_err(|e| IdxError::DatabaseError(e.to_string()))?; let tx_result: Result<(), IdxError> = (|| { let entity_id: i64 = match conn.query_row( "SELECT id FROM entities WHERE canonical_name = ?1", params![canonical_name], |row| row.get(0), ) { Ok(id) => id, Err(rusqlite::Error::QueryReturnedNoRows) => { let now = chrono::Utc::now().timestamp(); conn.execute( "INSERT INTO entities (canonical_name, entity_type, country, created_at, updated_at) VALUES (?1, NULL, NULL, ?2, ?2)", params![canonical_name, now], ) .map_err(|e| IdxError::DatabaseError(e.to_string()))?; conn.last_insert_rowid() } Err(e) => return Err(IdxError::DatabaseError(e.to_string())), }; for source in ["ksei", "bing"] { conn.execute( "INSERT INTO entity_aliases (entity_id, raw_name, source, confidence, method) VALUES (?1, ?2, ?3, 1.0, 'manual') ON CONFLICT(raw_name, source) DO UPDATE SET entity_id = excluded.entity_id, confidence = 1.0, method = 'manual'", params![entity_id, raw_name, source], ) .map_err(|e| IdxError::DatabaseError(e.to_string()))?; } conn.execute( "UPDATE ksei_holdings SET entity_id = ?1 WHERE raw_investor_name = ?2", params![entity_id, raw_name], ) .map_err(|e| IdxError::DatabaseError(e.to_string()))?; conn.execute( "UPDATE bing_holdings SET entity_id = ?1 WHERE raw_investor_name = ?2", params![entity_id, raw_name], ) .map_err(|e| IdxError::DatabaseError(e.to_string()))?; Ok(()) })(); match tx_result { Ok(()) => { conn.execute("COMMIT", []) .map_err(|e| IdxError::DatabaseError(e.to_string()))?; Ok(()) } Err(err) => { let _ = conn.execute("ROLLBACK", []); Err(err) } } } /// Merge two entities by re-pointing aliases and holdings from merge_id to keep_id, then deleting merge_id. pub fn merge_entities(conn: &Connection, keep_id: i64, merge_id: i64) -> Result<(), IdxError> { if keep_id == merge_id { return Err(IdxError::ParseError( "keep_id and merge_id must be different".to_string(), )); } conn.execute("BEGIN IMMEDIATE", []) .map_err(|e| IdxError::DatabaseError(e.to_string()))?; let tx_result: Result<(), IdxError> = (|| { let keep_exists: i64 = conn .query_row( "SELECT EXISTS(SELECT 1 FROM entities WHERE id = ?1)", params![keep_id], |row| row.get(0), ) .map_err(|e| IdxError::DatabaseError(e.to_string()))?; let merge_exists: i64 = conn .query_row( "SELECT EXISTS(SELECT 1 FROM entities WHERE id = ?1)", params![merge_id], |row| row.get(0), ) .map_err(|e| IdxError::DatabaseError(e.to_string()))?; if keep_exists != 1 || merge_exists != 1 { return Err(IdxError::ParseError(format!( "entity not found: keep_id={keep_id}, merge_id={merge_id}" ))); } conn.execute( "UPDATE entity_aliases SET entity_id = ?1 WHERE entity_id = ?2", params![keep_id, merge_id], ) .map_err(|e| IdxError::DatabaseError(e.to_string()))?; conn.execute( "UPDATE ksei_holdings SET entity_id = ?1 WHERE entity_id = ?2", params![keep_id, merge_id], ) .map_err(|e| IdxError::DatabaseError(e.to_string()))?; conn.execute( "UPDATE bing_holdings SET entity_id = ?1 WHERE entity_id = ?2", params![keep_id, merge_id], ) .map_err(|e| IdxError::DatabaseError(e.to_string()))?; conn.execute("DELETE FROM entities WHERE id = ?1", params![merge_id]) .map_err(|e| IdxError::DatabaseError(e.to_string()))?; Ok(()) })(); match tx_result { Ok(()) => { conn.execute("COMMIT", []) .map_err(|e| IdxError::DatabaseError(e.to_string()))?; Ok(()) } Err(err) => { let _ = conn.execute("ROLLBACK", []); Err(err) } } } /// Get ticker_id by code. pub fn get_ticker_id(conn: &Connection, code: &str) -> Result, IdxError> { conn.query_row( "SELECT id FROM tickers WHERE code = ?1", params![code], |row| row.get(0), ) .map(Some) .or_else(|e| match e { rusqlite::Error::QueryReturnedNoRows => Ok(None), _ => Err(IdxError::DatabaseError(e.to_string())), }) } /// Compute concentration metrics from a list of percentage_bps values. pub fn compute_concentration(percentages_bps: &[i64]) -> ConcentrationMetrics { let mut values = percentages_bps.to_vec(); values.sort_by(|a, b| b.cmp(a)); let top1_bps = values.first().copied().unwrap_or(0); let top3_bps: i64 = values.iter().take(3).sum(); let total_bps: i64 = values.iter().sum(); let hhi: i64 = values .iter() .map(|p| p.saturating_mul(*p) / 10000) .fold(0, i64::saturating_add); ConcentrationMetrics { top1_bps, top3_bps, hhi, free_float_bps: (10000 - total_bps).max(0), holder_count: values.iter().filter(|p| **p >= 100).count(), } } pub fn db_path() -> Result { if let Some(custom_path) = get_config_value("ownership.db_path")? { let trimmed = custom_path.trim(); if !trimmed.is_empty() { return Ok(PathBuf::from(trimmed)); } } if let Ok(dir) = std::env::var("XDG_DATA_HOME") && !dir.is_empty() { let path = PathBuf::from(dir); if path.is_absolute() { return Ok(path.join("idx").join("ownership.db")); } } ProjectDirs::from("", "", "idx") .map(|dirs| dirs.data_local_dir().join("ownership.db")) .ok_or_else(|| IdxError::DatabaseError("unable to resolve ownership db path".to_string())) } #[derive(Debug, Clone)] struct LatestKseiRelease { sha256: String, as_of_date: NaiveDate, } fn latest_ksei_release(conn: &Connection) -> Result, IdxError> { conn.query_row( "SELECT sha256, as_of_date FROM ownership_releases ORDER BY as_of_date DESC, imported_at DESC, id DESC LIMIT 1", [], |row| { Ok(LatestKseiRelease { sha256: row.get(0)?, as_of_date: parse_iso_date(&row.get::<_, String>(1)?) .map_err(|e| rusqlite::Error::ToSqlConversionFailure(Box::new(e)))?, }) }, ) .map(Some) .or_else(|e| match e { rusqlite::Error::QueryReturnedNoRows => Ok(None), _ => Err(IdxError::DatabaseError(e.to_string())), }) } fn query_ticker(conn: &Connection, code: &str) -> Result, IdxError> { conn.query_row( "SELECT id, code, name, sector FROM tickers WHERE code = ?1", params![code], |row| { Ok(Ticker { id: row.get(0)?, code: row.get(1)?, name: row.get(2)?, sector: row.get(3)?, }) }, ) .map(Some) .or_else(|e| match e { rusqlite::Error::QueryReturnedNoRows => Ok(None), _ => Err(IdxError::DatabaseError(e.to_string())), }) } fn query_entity(conn: &Connection, entity_id: i64) -> Result, IdxError> { conn.query_row( "SELECT id, canonical_name, entity_type, country FROM entities WHERE id = ?1", params![entity_id], |row| { Ok(Entity { id: row.get(0)?, canonical_name: row.get(1)?, entity_type: row.get(2)?, country: row.get(3)?, }) }, ) .map(Some) .or_else(|e| match e { rusqlite::Error::QueryReturnedNoRows => Ok(None), _ => Err(IdxError::DatabaseError(e.to_string())), }) } fn parse_iso_date(s: &str) -> Result { NaiveDate::parse_from_str(s, "%Y-%m-%d") .map_err(|e| IdxError::ParseError(format!("invalid ISO date '{s}': {e}"))) } fn locality_to_db(locality: Locality) -> &'static str { match locality { Locality::Local => "L", Locality::Foreign => "F", } } fn locality_from_db(value: &str) -> Option { match value { "L" => Some(Locality::Local), "F" | "A" => Some(Locality::Foreign), _ => None, } } fn flow_signal_to_db(signal: FlowSignal) -> &'static str { match signal { FlowSignal::Holder => "holder", FlowSignal::Buyer => "buyer", FlowSignal::Seller => "seller", FlowSignal::NewPosition => "new_position", FlowSignal::Exited => "exited", } } fn flow_signal_from_db(value: &str) -> Option { match value { "holder" => Some(FlowSignal::Holder), "buyer" => Some(FlowSignal::Buyer), "seller" => Some(FlowSignal::Seller), "new_position" => Some(FlowSignal::NewPosition), "exited" => Some(FlowSignal::Exited), _ => None, } } #[cfg(test)] mod tests { use chrono::NaiveDate; use rusqlite::{Connection, params}; use crate::error::IdxError; use crate::ownership::db::{ compute_concentration, ensure_schema, get_ticker_id, insert_bing_holdings, insert_ksei_holdings, insert_release, query_changes, query_concentration, query_cross_holders, query_ticker_holdings, release_exists, upsert_ticker, write_ksei_release, }; use crate::ownership::types::{ BingHolding, ChangeType, FlowSignal, KseiHolding, Locality, OwnershipRelease, }; fn setup() -> Connection { let conn = Connection::open_in_memory().unwrap(); ensure_schema(&conn).unwrap(); conn } /// Writes one release of `(entity_id, raw_name, bps)` rows for ticker_id. fn write_release( conn: &Connection, sha256: &str, date: &str, imported_at: i64, ticker_id: i64, rows: &[(Option, &str, i64)], ) { let as_of_date = NaiveDate::parse_from_str(date, "%Y-%m-%d").unwrap(); let holdings: Vec = rows .iter() .map(|&(entity_id, raw, bps)| KseiHolding { id: 0, ticker_id, entity_id, raw_investor_name: raw.to_string(), investor_type: None, locality: None, nationality: None, domicile: None, holdings_scripless: 1, holdings_scrip: 0, total_shares: 1, percentage_bps: bps, report_date: as_of_date, release_sha256: sha256.to_string(), }) .collect(); let release = OwnershipRelease { id: 0, source_url: None, sha256: sha256.to_string(), as_of_date, row_count: holdings.len(), imported_at, }; write_ksei_release(conn, &release, &holdings, false).unwrap(); } fn entity(conn: &Connection, name: &str) -> i64 { conn.execute( "INSERT INTO entities (canonical_name, entity_type, country, created_at, updated_at) VALUES (?1, NULL, NULL, 0, 0)", params![name], ) .unwrap(); conn.last_insert_rowid() } #[test] fn changes_sum_an_investors_accounts_in_a_ticker() { // KREN, 2026-09: a second account took KRESNA from 9.44% to 13.45%. let conn = setup(); let kren = upsert_ticker(&conn, "KREN", None).unwrap(); let kresna = entity(&conn, "KRESNA PRIMA INVEST"); write_release( &conn, "aug", "2026-08-31", 1, kren, &[(Some(kresna), "PT KRESNA PRIMA INVEST", 944)], ); write_release( &conn, "sep", "2026-09-30", 2, kren, &[ (Some(kresna), "PT KRESNA PRIMA INVEST", 944), (Some(kresna), "PT. KRESNA PRIMA INVEST", 401), ], ); let rows = query_changes(&conn, "2026-08-31", "2026-09-30").unwrap(); assert_eq!(rows.len(), 1, "{rows:?}"); assert_eq!(rows[0].change_type, ChangeType::Increased); assert_eq!( (rows[0].old_bps, rows[0].new_bps, rows[0].delta_bps), (Some(944), Some(1345), 401) ); } #[test] fn changes_read_the_latest_import_for_a_date() { // A corrected January file supersedes the original, so Jan -> Feb at // the corrected 12% shows no change, not a spurious increase. let conn = setup(); let bbca = upsert_ticker(&conn, "BBCA", None).unwrap(); let alpha = entity(&conn, "ALPHA"); write_release( &conn, "jan", "2026-01-30", 1, bbca, &[(Some(alpha), "ALPHA", 1000)], ); write_release( &conn, "jan-fix", "2026-01-30", 2, bbca, &[(Some(alpha), "ALPHA", 1200)], ); write_release( &conn, "feb", "2026-02-27", 3, bbca, &[(Some(alpha), "ALPHA", 1200)], ); assert!( query_changes(&conn, "2026-01-30", "2026-02-27") .unwrap() .is_empty() ); } #[test] fn changes_keep_each_date_to_its_own_rows() { // A file with a January ALPHA row and February BETA row is the // February release; ALPHA was never held in February, so Feb -> Mar // must not report it as exited. let conn = setup(); let bbca = upsert_ticker(&conn, "BBCA", None).unwrap(); write_release( &conn, "mixed", "2026-02-27", 1, bbca, &[(None, "ALPHA", 1000), (None, "BETA", 200)], ); conn.execute( "UPDATE ksei_holdings SET report_date = '2026-01-30' WHERE raw_investor_name = 'ALPHA'", [], ) .unwrap(); write_release(&conn, "mar", "2026-03-31", 2, bbca, &[(None, "BETA", 200)]); let rows = query_changes(&conn, "2026-02-27", "2026-03-31").unwrap(); assert!(rows.is_empty(), "{rows:?}"); let rows = query_changes(&conn, "2026-01-30", "2026-03-31").unwrap(); let alpha = rows.iter().find(|r| r.entity_name == "ALPHA").unwrap(); assert_eq!(alpha.change_type, ChangeType::Exited); } #[test] fn same_second_imports_resolve_to_the_later_file_everywhere() { // Original and corrected file for one date imported in the same // second: the ticker view and `changes` must read the same one. let conn = setup(); let bbca = upsert_ticker(&conn, "BBCA", None).unwrap(); let alpha = entity(&conn, "ALPHA"); write_release( &conn, "jan", "2026-01-30", 7, bbca, &[(Some(alpha), "ALPHA", 1000)], ); write_release( &conn, "jan-fix", "2026-01-30", 7, bbca, &[(Some(alpha), "ALPHA", 1200)], ); let ticker = query_ticker_holdings(&conn, "BBCA").unwrap(); assert_eq!( ticker.holders[0].percentage_bps, 1200, "{:?}", ticker.holders ); } #[test] fn a_release_records_the_rows_it_stored() { // A file listing the same ticker/investor twice stores one row; the // release metadata must say 1, or snapshot validation rejects it. let conn = setup(); let bbca = upsert_ticker(&conn, "BBCA", None).unwrap(); write_release( &conn, "dup", "2026-02-27", 1, bbca, &[(None, "ALPHA", 1000), (None, "ALPHA", 1000)], ); let row_count: i64 = conn .query_row( "SELECT row_count FROM ownership_releases WHERE sha256 = 'dup'", [], |row| row.get(0), ) .unwrap(); let stored: i64 = conn .query_row( "SELECT COUNT(*) FROM ksei_holdings WHERE release_sha256 = 'dup'", [], |row| row.get(0), ) .unwrap(); assert_eq!((row_count, stored), (1, 1)); } #[test] fn changes_reject_a_date_without_a_release() { let conn = setup(); let bbca = upsert_ticker(&conn, "BBCA", None).unwrap(); write_release( &conn, "feb", "2026-02-27", 1, bbca, &[(None, "ALPHA", 1200)], ); let err = query_changes(&conn, "2026-01-30", "2026-02-27").unwrap_err(); assert!( matches!(err, IdxError::InvalidInput(ref m) if m.contains("2026-01-30")), "{err:?}" ); } #[test] fn test_compute_concentration_pure_function() { let metrics = compute_concentration(&[4000, 2500, 1000, 500]); assert_eq!(metrics.top1_bps, 4000); assert_eq!(metrics.top3_bps, 7500); assert_eq!(metrics.hhi, 2350); assert_eq!(metrics.free_float_bps, 2000); assert_eq!(metrics.holder_count, 4); } #[test] fn test_release_exists_before_and_after_insert() { let conn = setup(); assert!(!release_exists(&conn, "abc").unwrap()); let release = OwnershipRelease { id: 0, source_url: Some("https://example.com/r.pdf".to_string()), sha256: "abc".to_string(), as_of_date: NaiveDate::from_ymd_opt(2026, 2, 27).unwrap(), row_count: 10, imported_at: 1_700_000_000, }; let _ = insert_release(&conn, &release).unwrap(); assert!(release_exists(&conn, "abc").unwrap()); } #[test] fn test_insert_fixture_ksei_and_query_ticker() { let conn = setup(); let bbca_id = upsert_ticker(&conn, "BBCA", Some("BCA")).unwrap(); conn.execute( "INSERT INTO entities (canonical_name, entity_type, country, created_at, updated_at) VALUES ('ALPHA FUND', NULL, NULL, 0, 0)", [], ) .unwrap(); let alpha_id = conn.last_insert_rowid(); let holdings = vec![ KseiHolding { id: 0, ticker_id: bbca_id, entity_id: Some(alpha_id), raw_investor_name: "PT ALPHA FUND".to_string(), investor_type: None, locality: Some(Locality::Local), nationality: None, domicile: None, holdings_scripless: 1_000, holdings_scrip: 0, total_shares: 1_000, percentage_bps: 4000, report_date: NaiveDate::from_ymd_opt(2026, 2, 27).unwrap(), release_sha256: "r1".to_string(), }, KseiHolding { id: 0, ticker_id: bbca_id, entity_id: None, raw_investor_name: "BETA".to_string(), investor_type: None, locality: Some(Locality::Foreign), nationality: None, domicile: None, holdings_scripless: 600, holdings_scrip: 0, total_shares: 600, percentage_bps: 2400, report_date: NaiveDate::from_ymd_opt(2026, 2, 27).unwrap(), release_sha256: "r1".to_string(), }, ]; let release = OwnershipRelease { id: 0, source_url: None, sha256: "r1".to_string(), as_of_date: NaiveDate::from_ymd_opt(2026, 2, 27).unwrap(), row_count: holdings.len(), imported_at: 1, }; assert_eq!( write_ksei_release(&conn, &release, &holdings, false).unwrap(), 2 ); let data = query_ticker_holdings(&conn, "BBCA").unwrap(); assert_eq!(data.ticker.code, "BBCA"); assert_eq!(data.holders.len(), 2); assert_eq!(data.holders[0].percentage_bps, 4000); assert_eq!(data.concentration.top1_bps, 4000); assert_eq!(data.concentration.top3_bps, 6400); } #[test] fn test_duplicate_insert_ignored() { let conn = setup(); let bbri_id = upsert_ticker(&conn, "BBRI", None).unwrap(); let h = KseiHolding { id: 0, ticker_id: bbri_id, entity_id: None, raw_investor_name: "DUP".to_string(), investor_type: None, locality: None, nationality: None, domicile: None, holdings_scripless: 10, holdings_scrip: 0, total_shares: 10, percentage_bps: 100, report_date: NaiveDate::from_ymd_opt(2026, 2, 27).unwrap(), release_sha256: "same-release".to_string(), }; assert_eq!( insert_ksei_holdings(&conn, std::slice::from_ref(&h)).unwrap(), 1 ); assert_eq!(insert_ksei_holdings(&conn, &[h]).unwrap(), 0); let count: i64 = conn .query_row("SELECT COUNT(*) FROM ksei_holdings", [], |row| row.get(0)) .unwrap(); assert_eq!(count, 1); } #[test] fn test_write_ksei_release_replaces_existing_sha_when_requested() { let conn = setup(); let bbri_id = upsert_ticker(&conn, "BBRI", None).unwrap(); let initial = KseiHolding { id: 0, ticker_id: bbri_id, entity_id: None, raw_investor_name: "FORCE".to_string(), investor_type: None, locality: None, nationality: None, domicile: None, holdings_scripless: 10, holdings_scrip: 0, total_shares: 10, percentage_bps: 100, report_date: NaiveDate::from_ymd_opt(2026, 2, 27).unwrap(), release_sha256: "force-sha".to_string(), }; let initial_release = OwnershipRelease { id: 0, source_url: None, sha256: "force-sha".to_string(), as_of_date: NaiveDate::from_ymd_opt(2026, 2, 27).unwrap(), row_count: 1, imported_at: 1, }; assert_eq!( write_ksei_release( &conn, &initial_release, std::slice::from_ref(&initial), false ) .unwrap(), 1 ); let replacement = KseiHolding { percentage_bps: 250, total_shares: 25, holdings_scripless: 25, ..initial }; let replacement_release = OwnershipRelease { imported_at: 2, ..initial_release }; assert_eq!( write_ksei_release( &conn, &replacement_release, std::slice::from_ref(&replacement), true, ) .unwrap(), 1 ); let release_count: i64 = conn .query_row( "SELECT COUNT(*) FROM ownership_releases WHERE sha256 = 'force-sha'", [], |row| row.get(0), ) .unwrap(); assert_eq!(release_count, 1); let latest_bps: i64 = conn .query_row( "SELECT percentage_bps FROM ksei_holdings WHERE release_sha256 = 'force-sha'", [], |row| row.get(0), ) .unwrap(); assert_eq!(latest_bps, 250); } #[test] fn test_query_ticker_holdings_uses_latest_release_only() { let conn = setup(); let tlkm_id = upsert_ticker(&conn, "TLKM", None).unwrap(); let older_release = OwnershipRelease { id: 0, source_url: None, sha256: "old-release".to_string(), as_of_date: NaiveDate::from_ymd_opt(2026, 1, 30).unwrap(), row_count: 2, imported_at: 1, }; let older_rows = vec![ KseiHolding { id: 0, ticker_id: tlkm_id, entity_id: None, raw_investor_name: "OLDER A".to_string(), investor_type: None, locality: None, nationality: None, domicile: None, holdings_scripless: 40, holdings_scrip: 0, total_shares: 40, percentage_bps: 4000, report_date: older_release.as_of_date, release_sha256: older_release.sha256.clone(), }, KseiHolding { id: 0, ticker_id: tlkm_id, entity_id: None, raw_investor_name: "OLDER B".to_string(), investor_type: None, locality: None, nationality: None, domicile: None, holdings_scripless: 30, holdings_scrip: 0, total_shares: 30, percentage_bps: 3000, report_date: older_release.as_of_date, release_sha256: older_release.sha256.clone(), }, ]; write_ksei_release(&conn, &older_release, &older_rows, false).unwrap(); let latest_release = OwnershipRelease { id: 0, source_url: None, sha256: "latest-release".to_string(), as_of_date: NaiveDate::from_ymd_opt(2026, 2, 27).unwrap(), row_count: 1, imported_at: 2, }; let latest_rows = vec![KseiHolding { id: 0, ticker_id: tlkm_id, entity_id: None, raw_investor_name: "LATEST".to_string(), investor_type: None, locality: None, nationality: None, domicile: None, holdings_scripless: 25, holdings_scrip: 0, total_shares: 25, percentage_bps: 2500, report_date: latest_release.as_of_date, release_sha256: latest_release.sha256.clone(), }]; write_ksei_release(&conn, &latest_release, &latest_rows, false).unwrap(); let data = query_ticker_holdings(&conn, "TLKM").unwrap(); assert_eq!(data.ksei_as_of, Some(latest_release.as_of_date)); assert_eq!(data.holders.len(), 1); assert_eq!(data.holders[0].name, "LATEST"); assert_eq!(data.concentration.top1_bps, 2500); } #[test] fn test_cross_holders_same_entity_three_tickers() { let conn = setup(); conn.execute( "INSERT INTO entities (canonical_name, entity_type, country, created_at, updated_at) VALUES ('OMEGA', NULL, NULL, 0, 0)", [], ) .unwrap(); let eid = conn.last_insert_rowid(); let mut holdings = Vec::new(); for (code, bps) in [("BBCA", 1000), ("BBRI", 1100), ("BMRI", 1200)] { let tid = upsert_ticker(&conn, code, None).unwrap(); holdings.push(KseiHolding { id: 0, ticker_id: tid, entity_id: Some(eid), raw_investor_name: "OMEGA".to_string(), investor_type: None, locality: None, nationality: None, domicile: None, holdings_scripless: 10, holdings_scrip: 0, total_shares: 10, percentage_bps: bps, report_date: NaiveDate::from_ymd_opt(2026, 2, 27).unwrap(), release_sha256: "current-release".to_string(), }); } let release = OwnershipRelease { id: 0, source_url: None, sha256: "current-release".to_string(), as_of_date: NaiveDate::from_ymd_opt(2026, 2, 27).unwrap(), row_count: holdings.len(), imported_at: 1, }; write_ksei_release(&conn, &release, &holdings, false).unwrap(); let rows = query_cross_holders(&conn, 3, 10).unwrap(); assert_eq!(rows.len(), 1); assert_eq!(rows[0].entity.id, eid); assert_eq!(rows[0].ticker_count, 3); } #[test] fn test_cross_holders_use_latest_release_only() { let conn = setup(); conn.execute( "INSERT INTO entities (canonical_name, entity_type, country, created_at, updated_at) VALUES ('OMEGA', NULL, NULL, 0, 0)", [], ) .unwrap(); let eid = conn.last_insert_rowid(); let aa = upsert_ticker(&conn, "AA", None).unwrap(); let bb = upsert_ticker(&conn, "BB", None).unwrap(); let old_release = OwnershipRelease { id: 0, source_url: None, sha256: "old-cross".to_string(), as_of_date: NaiveDate::from_ymd_opt(2026, 1, 30).unwrap(), row_count: 2, imported_at: 1, }; let old_rows = vec![ KseiHolding { id: 0, ticker_id: aa, entity_id: Some(eid), raw_investor_name: "OMEGA".to_string(), investor_type: None, locality: None, nationality: None, domicile: None, holdings_scripless: 1, holdings_scrip: 0, total_shares: 1, percentage_bps: 1000, report_date: old_release.as_of_date, release_sha256: old_release.sha256.clone(), }, KseiHolding { id: 0, ticker_id: bb, entity_id: Some(eid), raw_investor_name: "OMEGA".to_string(), investor_type: None, locality: None, nationality: None, domicile: None, holdings_scripless: 1, holdings_scrip: 0, total_shares: 1, percentage_bps: 1200, report_date: old_release.as_of_date, release_sha256: old_release.sha256.clone(), }, ]; write_ksei_release(&conn, &old_release, &old_rows, false).unwrap(); let latest_release = OwnershipRelease { id: 0, source_url: None, sha256: "latest-cross".to_string(), as_of_date: NaiveDate::from_ymd_opt(2026, 2, 27).unwrap(), row_count: 1, imported_at: 2, }; let latest_rows = vec![KseiHolding { id: 0, ticker_id: aa, entity_id: Some(eid), raw_investor_name: "OMEGA".to_string(), investor_type: None, locality: None, nationality: None, domicile: None, holdings_scripless: 1, holdings_scrip: 0, total_shares: 1, percentage_bps: 900, report_date: latest_release.as_of_date, release_sha256: latest_release.sha256.clone(), }]; write_ksei_release(&conn, &latest_release, &latest_rows, false).unwrap(); assert!(query_cross_holders(&conn, 2, 10).unwrap().is_empty()); let rows = query_cross_holders(&conn, 1, 10).unwrap(); assert_eq!(rows.len(), 1); assert_eq!(rows[0].ticker_count, 1); assert_eq!(rows[0].total_bps, 900); } #[test] fn test_concentration_sorting() { let conn = setup(); let aa = upsert_ticker(&conn, "AA", None).unwrap(); let bb = upsert_ticker(&conn, "BB", None).unwrap(); let rows = vec![ (aa, 6000, "A1"), (aa, 1000, "A2"), (bb, 5000, "B1"), (bb, 3000, "B2"), ]; let holdings = rows .into_iter() .map(|(tid, bps, name)| KseiHolding { id: 0, ticker_id: tid, entity_id: None, raw_investor_name: name.to_string(), investor_type: None, locality: None, nationality: None, domicile: None, holdings_scripless: 1, holdings_scrip: 0, total_shares: 1, percentage_bps: bps, report_date: NaiveDate::from_ymd_opt(2026, 2, 27).unwrap(), release_sha256: "current-release".to_string(), }) .collect::>(); let release = OwnershipRelease { id: 0, source_url: None, sha256: "current-release".to_string(), as_of_date: NaiveDate::from_ymd_opt(2026, 2, 27).unwrap(), row_count: holdings.len(), imported_at: 1, }; write_ksei_release(&conn, &release, &holdings, false).unwrap(); let by_top1 = query_concentration(&conn, "top1", 10).unwrap(); assert_eq!(by_top1[0].0, "AA"); let by_top3 = query_concentration(&conn, "top3", 10).unwrap(); assert_eq!(by_top3[0].0, "BB"); let by_hhi = query_concentration(&conn, "hhi", 10).unwrap(); assert_eq!(by_hhi[0].0, "AA"); } #[test] fn test_concentration_uses_latest_release_only() { let conn = setup(); let aa = upsert_ticker(&conn, "AA", None).unwrap(); let old_release = OwnershipRelease { id: 0, source_url: None, sha256: "old-concentration".to_string(), as_of_date: NaiveDate::from_ymd_opt(2026, 1, 30).unwrap(), row_count: 1, imported_at: 1, }; let old_rows = vec![KseiHolding { id: 0, ticker_id: aa, entity_id: None, raw_investor_name: "OLD".to_string(), investor_type: None, locality: None, nationality: None, domicile: None, holdings_scripless: 1, holdings_scrip: 0, total_shares: 1, percentage_bps: 6000, report_date: old_release.as_of_date, release_sha256: old_release.sha256.clone(), }]; write_ksei_release(&conn, &old_release, &old_rows, false).unwrap(); let latest_release = OwnershipRelease { id: 0, source_url: None, sha256: "latest-concentration".to_string(), as_of_date: NaiveDate::from_ymd_opt(2026, 2, 27).unwrap(), row_count: 2, imported_at: 2, }; let latest_rows = vec![ KseiHolding { id: 0, ticker_id: aa, entity_id: None, raw_investor_name: "NEW A".to_string(), investor_type: None, locality: None, nationality: None, domicile: None, holdings_scripless: 1, holdings_scrip: 0, total_shares: 1, percentage_bps: 2000, report_date: latest_release.as_of_date, release_sha256: latest_release.sha256.clone(), }, KseiHolding { id: 0, ticker_id: aa, entity_id: None, raw_investor_name: "NEW B".to_string(), investor_type: None, locality: None, nationality: None, domicile: None, holdings_scripless: 1, holdings_scrip: 0, total_shares: 1, percentage_bps: 1500, report_date: latest_release.as_of_date, release_sha256: latest_release.sha256.clone(), }, ]; write_ksei_release(&conn, &latest_release, &latest_rows, false).unwrap(); let rows = query_concentration(&conn, "top1", 10).unwrap(); assert_eq!(rows.len(), 1); assert_eq!(rows[0].1.top1_bps, 2000); assert_eq!(rows[0].1.top3_bps, 3500); } #[test] fn test_insert_bing_and_query_empty_db_paths() { let conn = setup(); assert!(get_ticker_id(&conn, "NONE").unwrap().is_none()); assert!(query_cross_holders(&conn, 1, 10).unwrap().is_empty()); assert!(query_concentration(&conn, "top1", 10).unwrap().is_empty()); let tid = upsert_ticker(&conn, "TLKM", None).unwrap(); let inserted = insert_bing_holdings( &conn, &[BingHolding { id: 0, ticker_id: tid, entity_id: None, raw_investor_name: "FLOW".to_string(), investor_type: None, shares_held: Some(100), shares_changed: Some(50), pct_ownership_bps: Some(123), value_usd: Some(10), report_date: NaiveDate::from_ymd_opt(2026, 2, 27).unwrap(), signal: FlowSignal::Buyer, fetched_at: 12345, }], ) .unwrap(); assert_eq!(inserted, 1); let buyers: i64 = conn .query_row( "SELECT COUNT(*) FROM bing_holdings WHERE ticker_id = ?1 AND signal = 'buyer'", params![tid], |row| row.get(0), ) .unwrap(); assert_eq!(buyers, 1); } #[cfg(unix)] #[test] fn resolve_db_path_follows_symlinks_even_when_the_target_is_missing() { let root = std::env::temp_dir().join(format!( "idx-resolve-{}-{}", std::process::id(), fastrand::u64(..) )); let real_dir = root.join("real"); std::fs::create_dir_all(&real_dir).unwrap(); let real_dir = std::fs::canonicalize(&real_dir).unwrap(); let link = root.join("link.db"); // Dangling: the target does not exist yet (fresh install, mid-swap). std::os::unix::fs::symlink(real_dir.join("ownership.db"), &link).unwrap(); assert_eq!( super::resolve_db_path(&link).unwrap(), real_dir.join("ownership.db") ); // Relative link, resolved against the link's directory. let relative = root.join("relative.db"); std::os::unix::fs::symlink("real/ownership.db", &relative).unwrap(); assert_eq!( super::resolve_db_path(&relative).unwrap(), real_dir.join("ownership.db") ); // A plain path is returned as-is (canonical parent). assert_eq!( super::resolve_db_path(&real_dir.join("ownership.db")).unwrap(), real_dir.join("ownership.db") ); let _ = std::fs::remove_dir_all(&root); } }