refactor: split into idx-core library and idx-cli binary workspace

Phase 2 of docs/DESIGN.md, no behavior change:

- crates/idx-core (library): api, analysis, cache, config file handling
  and provider selection, curl_impersonate, error, ownership, runtime,
  plus the shared test fixtures. clap derives are behind an optional
  `clap` feature so the library does not require clap.
- crates/idx-cli (binary `idx`): cli/, output/, and IdxConfig (output
  format and color resolution). Package and binary names are unchanged.
- config.rs is split: file read/write and validation move to core; a
  CLI test keeps core's OUTPUT_VALUES in sync with OutputSetting.
- ownership::db::open_db/db_path drop their unused IdxConfig argument.
- One workspace version; shared dependency versions; no third-party
  version changes in Cargo.lock. LICENSE is symlinked into each crate.
- flake: workspace version, new source paths, and a per-crate package
  check (idx-cli is built from its tarball against packaged idx-core,
  since idx-core is not on crates.io yet). CI cache key, release path
  filter, and the golden whitespace-hook exclusion follow the new layout.

Parity: all 143 golden outputs match byte for byte without
regeneration; 157 core + 17 CLI unit tests (173 before plus the new
OUTPUT_VALUES sync test), 62 CLI tests pass. Live A/B against main:
16/16 stocks cases identical; ownership discover got IDX 403 on both
(exit 4 vs 1 is the phase 1 exit-code change).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
hermes 2026-10-14 09:13:46 +00:00
commit 15947de51a
238 changed files with 544 additions and 375 deletions

View file

@ -0,0 +1,360 @@
use std::fs;
use std::io::{Cursor, Read};
use std::path::Path;
use chrono::NaiveDate;
use crate::error::IdxError;
use crate::ownership::types::{InvestorTypeCode, KseiHoldingDraft, Locality};
const EXPECTED_HEADER: &[&str] = &[
"Date",
"Code",
"Type",
"Sec. Num",
"Price",
"Local IS",
"Local CP",
"Local PF",
"Local IB",
"Local ID",
"Local MF",
"Local SC",
"Local FD",
"Local OT",
"Total",
"Foreign IS",
"Foreign CP",
"Foreign PF",
"Foreign IB",
"Foreign ID",
"Foreign MF",
"Foreign SC",
"Foreign FD",
"Foreign OT",
"Total",
];
const LOCAL_BUCKETS: &[(&str, usize)] = &[
("IS", 5),
("CP", 6),
("PF", 7),
("IB", 8),
("ID", 9),
("MF", 10),
("SC", 11),
("FD", 12),
("OT", 13),
];
const FOREIGN_BUCKETS: &[(&str, usize)] = &[
("IS", 15),
("CP", 16),
("PF", 17),
("IB", 18),
("ID", 19),
("MF", 20),
("SC", 21),
("FD", 22),
("OT", 23),
];
pub fn supports_local_archive_file(path: &Path) -> bool {
path.extension()
.and_then(|value| value.to_str())
.map(|value| {
let ext = value.trim().to_ascii_lowercase();
ext == "txt" || ext == "zip"
})
.unwrap_or(false)
}
pub fn parse_balancepos_file(path: &Path) -> Result<Vec<KseiHoldingDraft>, IdxError> {
let raw = match path
.extension()
.and_then(|value| value.to_str())
.map(|value| value.trim().to_ascii_lowercase())
.as_deref()
{
Some("txt") => fs::read_to_string(path).map_err(|e| {
IdxError::Io(format!(
"failed to read KSEI archive TXT {}: {e}",
path.display()
))
})?,
Some("zip") => extract_txt_from_zip(path)?,
_ => {
return Err(IdxError::InvalidInput(format!(
"unsupported local ownership archive file {}; expected .txt or .zip",
path.display()
)));
}
};
parse_balancepos_text(&raw)
}
pub fn parse_balancepos_text(raw: &str) -> Result<Vec<KseiHoldingDraft>, IdxError> {
let mut lines = raw.lines().filter(|line| !line.trim().is_empty());
let header = lines
.next()
.ok_or_else(|| IdxError::ParseError("empty KSEI archive TXT input".to_string()))?;
validate_header(header)?;
let mut drafts = Vec::new();
for (line_number, line) in lines.enumerate() {
let columns: Vec<&str> = line.split('|').map(str::trim).collect();
if columns.len() != EXPECTED_HEADER.len() {
return Err(IdxError::ParseError(format!(
"invalid KSEI archive TXT row {}: expected {} columns, got {}",
line_number + 2,
EXPECTED_HEADER.len(),
columns.len()
)));
}
if !columns[2].eq_ignore_ascii_case("EQUITY") {
continue;
}
let report_date = parse_archive_date(columns[0])?;
let ticker_code = columns[1].trim().to_uppercase();
let sec_num = parse_archive_number(columns[3], line_number + 2, "Sec. Num")?;
if sec_num <= 0 {
continue;
}
append_bucket_drafts(
&mut drafts,
&ticker_code,
report_date,
sec_num,
columns.as_slice(),
Locality::Local,
LOCAL_BUCKETS,
)?;
append_bucket_drafts(
&mut drafts,
&ticker_code,
report_date,
sec_num,
columns.as_slice(),
Locality::Foreign,
FOREIGN_BUCKETS,
)?;
}
if drafts.is_empty() {
return Err(IdxError::ParseError(
"no importable EQUITY rows found in KSEI archive TXT".to_string(),
));
}
Ok(drafts)
}
fn extract_txt_from_zip(path: &Path) -> Result<String, IdxError> {
let bytes = fs::read(path).map_err(|e| {
IdxError::Io(format!(
"failed to read KSEI archive ZIP {}: {e}",
path.display()
))
})?;
let cursor = Cursor::new(bytes);
let mut zip = zip::ZipArchive::new(cursor)
.map_err(|e| IdxError::ParseError(format!("failed to open KSEI archive ZIP: {e}")))?;
for index in 0..zip.len() {
let mut file = zip.by_index(index).map_err(|e| {
IdxError::ParseError(format!("failed to read KSEI archive ZIP entry: {e}"))
})?;
if file.is_dir() {
continue;
}
let name = file.name().to_ascii_lowercase();
if !name.ends_with(".txt") {
continue;
}
let mut output = String::new();
file.read_to_string(&mut output).map_err(|e| {
IdxError::ParseError(format!("failed to decode KSEI archive TXT entry: {e}"))
})?;
return Ok(output);
}
Err(IdxError::ParseError(
"KSEI archive ZIP did not contain a TXT payload".to_string(),
))
}
fn validate_header(header: &str) -> Result<(), IdxError> {
let columns: Vec<&str> = header.split('|').map(str::trim).collect();
if columns != EXPECTED_HEADER {
return Err(IdxError::ParseError(
"KSEI archive TXT header did not match the expected balancepos layout".to_string(),
));
}
Ok(())
}
fn parse_archive_date(raw: &str) -> Result<NaiveDate, IdxError> {
let trimmed = raw.trim();
if trimmed.len() != 11 {
return Err(IdxError::ParseError(format!(
"invalid KSEI archive date `{trimmed}`"
)));
}
let canonical = format!(
"{}-{}-{}",
&trimmed[..2],
titlecase_month(&trimmed[3..6]),
&trimmed[7..11]
);
NaiveDate::parse_from_str(&canonical, "%d-%b-%Y")
.map_err(|e| IdxError::ParseError(format!("invalid KSEI archive date `{trimmed}`: {e}")))
}
fn titlecase_month(raw: &str) -> String {
let upper = raw.trim().to_ascii_uppercase();
let mut chars = upper.chars();
match chars.next() {
Some(first) => {
let mut output = String::new();
output.push(first.to_ascii_uppercase());
output.push_str(&chars.as_str().to_ascii_lowercase());
output
}
None => String::new(),
}
}
fn parse_archive_number(raw: &str, line_number: usize, field: &str) -> Result<i64, IdxError> {
raw.trim().parse::<i64>().map_err(|e| {
IdxError::ParseError(format!(
"invalid KSEI archive TXT value in row {line_number} field `{field}`: {e}"
))
})
}
fn append_bucket_drafts(
drafts: &mut Vec<KseiHoldingDraft>,
ticker_code: &str,
report_date: NaiveDate,
sec_num: i64,
columns: &[&str],
locality: Locality,
buckets: &[(&str, usize)],
) -> Result<(), IdxError> {
for (investor_type, column_index) in buckets {
let shares =
parse_archive_number(columns[*column_index], 0, investor_type).map_err(|_| {
IdxError::ParseError(format!(
"invalid KSEI archive TXT share count for {ticker_code} {investor_type}"
))
})?;
if shares <= 0 {
continue;
}
drafts.push(KseiHoldingDraft {
ticker_code: ticker_code.to_string(),
issuer_name: None,
raw_investor_name: synthetic_holder_name(locality, investor_type),
investor_type: Some(InvestorTypeCode((*investor_type).to_string())),
locality: Some(locality),
nationality: None,
domicile: None,
holdings_scripless: shares,
holdings_scrip: 0,
total_shares: shares,
percentage_bps: compute_percentage_bps(shares, sec_num),
report_date,
});
}
Ok(())
}
fn synthetic_holder_name(locality: Locality, investor_type: &str) -> String {
let prefix = match locality {
Locality::Local => "LOCAL",
Locality::Foreign => "FOREIGN",
};
format!("KSEI AGGREGATE {prefix} {investor_type}")
}
fn compute_percentage_bps(shares: i64, sec_num: i64) -> i64 {
let shares_i128 = i128::from(shares);
let sec_num_i128 = i128::from(sec_num);
let rounded = ((shares_i128 * 10_000) + (sec_num_i128 / 2)) / sec_num_i128;
i64::try_from(rounded).unwrap_or(0)
}
#[cfg(test)]
mod tests {
use super::parse_balancepos_text;
use crate::ownership::entities::normalize_ksei_row;
use crate::ownership::parser::parse_stext_xml;
use crate::ownership::types::Locality;
#[test]
fn parses_balancepos_excerpt_into_bucket_holders() {
let raw = include_str!("../../tests/fixtures/ksei_balancepos_20260227_excerpt.txt");
let drafts = parse_balancepos_text(raw).expect("balancepos excerpt parses");
let local_cp = drafts
.iter()
.find(|draft| {
draft.ticker_code == "AADI"
&& draft.investor_type.as_ref().map(|code| code.0.as_str()) == Some("CP")
&& draft.locality == Some(Locality::Local)
})
.expect("AADI local CP bucket");
assert_eq!(local_cp.raw_investor_name, "KSEI AGGREGATE LOCAL CP");
assert_eq!(local_cp.total_shares, 5_035_745_466);
assert_eq!(local_cp.percentage_bps, 6467);
}
#[test]
fn balancepos_cross_check_contains_pdf_holder_bucket() {
let pdf_rows = parse_stext_xml(include_str!(
"../../tests/fixtures/ksei_above1_stext_excerpt.xml"
))
.expect("pdf fixture rows");
let pdf_drafts: Vec<_> = pdf_rows
.iter()
.map(normalize_ksei_row)
.collect::<Result<_, _>>()
.expect("normalized pdf drafts");
let archive_drafts = parse_balancepos_text(include_str!(
"../../tests/fixtures/ksei_balancepos_20260227_excerpt.txt"
))
.expect("archive drafts");
let pdf_local_cp = pdf_drafts
.iter()
.find(|draft| {
draft.ticker_code == "AADI"
&& draft.investor_type.as_ref().map(|code| code.0.as_str()) == Some("CP")
&& draft.locality == Some(Locality::Local)
})
.expect("pdf local cp");
let archive_local_cp = archive_drafts
.iter()
.find(|draft| {
draft.ticker_code == "AADI"
&& draft.investor_type.as_ref().map(|code| code.0.as_str()) == Some("CP")
&& draft.locality == Some(Locality::Local)
})
.expect("archive local cp");
assert_eq!(pdf_local_cp.report_date, archive_local_cp.report_date);
assert!(pdf_local_cp.total_shares <= archive_local_cp.total_shares);
assert!(pdf_local_cp.percentage_bps <= archive_local_cp.percentage_bps);
}
}

File diff suppressed because it is too large Load diff

View file

@ -0,0 +1,456 @@
use chrono::NaiveDate;
use rusqlite::{Connection, params};
use crate::error::IdxError;
use crate::ownership::types::{
InvestorTypeCode, KseiHoldingDraft, KseiRawRow, Locality, OwnershipSource,
};
/// Parse Indonesian locale number string to i64.
/// "1.533.682.440" → 1533682440, "0" → 0
pub fn parse_id_number(s: &str) -> Result<i64, IdxError> {
let trimmed = s.trim();
if trimmed.is_empty() {
return Err(IdxError::ParseError(
"empty Indonesian number string".to_string(),
));
}
let normalized = trimmed.replace(['.', ' '], "");
if normalized.is_empty() || !normalized.chars().all(|c| c.is_ascii_digit()) {
return Err(IdxError::ParseError(format!(
"invalid Indonesian number format: {s}"
)));
}
normalized
.parse::<i64>()
.map_err(|e| IdxError::ParseError(format!("failed to parse Indonesian number '{s}': {e}")))
}
fn parse_id_number_or_zero(s: &str) -> Result<i64, IdxError> {
if s.trim().is_empty() {
Ok(0)
} else {
parse_id_number(s)
}
}
/// Parse Indonesian locale percentage to basis points (i64).
/// "54,94" → 5494, "0,00" → 0, "100,00" → 10000
pub fn parse_id_percentage(s: &str) -> Result<i64, IdxError> {
let trimmed = s.trim();
if trimmed.is_empty() {
return Err(IdxError::ParseError(
"empty Indonesian percentage string".to_string(),
));
}
let mut parts = trimmed.split(',');
let whole = parts.next().unwrap_or_default();
let frac = parts.next().unwrap_or("00");
if parts.next().is_some() {
return Err(IdxError::ParseError(format!(
"invalid Indonesian percentage format: {s}"
)));
}
if whole.is_empty() || !whole.chars().all(|c| c.is_ascii_digit()) {
return Err(IdxError::ParseError(format!(
"invalid Indonesian percentage whole part: {s}"
)));
}
if !frac.chars().all(|c| c.is_ascii_digit()) || frac.len() > 2 {
return Err(IdxError::ParseError(format!(
"invalid Indonesian percentage fractional part: {s}"
)));
}
let whole_i = whole.parse::<i64>().map_err(|e| {
IdxError::ParseError(format!("failed to parse Indonesian percentage '{s}': {e}"))
})?;
let frac_padded = if frac.len() == 1 {
format!("{frac}0")
} else {
frac.to_string()
};
let frac_i = frac_padded.parse::<i64>().map_err(|e| {
IdxError::ParseError(format!("failed to parse Indonesian percentage '{s}': {e}"))
})?;
Ok((whole_i * 100) + frac_i)
}
/// Parse KSEI date format to NaiveDate.
/// "27-Feb-2026" → NaiveDate(2026, 2, 27)
pub fn parse_ksei_date(s: &str) -> Result<NaiveDate, IdxError> {
let trimmed = s.trim();
if trimmed.is_empty() {
return Err(IdxError::ParseError("empty KSEI date string".to_string()));
}
NaiveDate::parse_from_str(trimmed, "%d-%b-%Y")
.map_err(|e| IdxError::ParseError(format!("invalid KSEI date '{s}': {e}")))
}
/// Normalize an investor name for entity matching.
/// Strips: "PT.", "PT ", "TBK", "Tbk", "(PERSERO)", "LIMITED", "PTE", "LTD"
/// Collapses whitespace, trims, uppercases.
pub fn normalize_name(raw: &str) -> String {
let upper = collapse_whitespace(raw).to_uppercase();
if upper.is_empty() {
return upper;
}
let mut tokens: Vec<&str> = upper.split_whitespace().collect();
loop {
let mut changed = false;
if let Some(first) = tokens.first().copied()
&& matches!(first, "PT" | "PT.")
{
tokens.remove(0);
changed = true;
}
if let Some(last) = tokens.last().copied()
&& matches!(
last,
"TBK" | "Tbk" | "(PERSERO)" | "LIMITED" | "PTE" | "LTD"
)
{
let _ = tokens.pop();
changed = true;
}
if !changed {
break;
}
}
collapse_whitespace(&tokens.join(" "))
}
/// Convert a KseiRawRow into a normalized KseiHoldingDraft.
/// Applies all parsing functions above.
pub fn normalize_ksei_row(raw: &KseiRawRow) -> Result<KseiHoldingDraft, IdxError> {
let investor_type = normalize_investor_type(&raw.investor_type);
let locality = normalize_locality(&raw.local_foreign);
Ok(KseiHoldingDraft {
ticker_code: raw.share_code.trim().to_string(),
issuer_name: optional_string(&raw.issuer_name),
raw_investor_name: raw.investor_name.trim().to_string(),
investor_type,
locality,
nationality: optional_string(&raw.nationality),
domicile: optional_string(&raw.domicile),
holdings_scripless: parse_id_number_or_zero(&raw.holdings_scripless)?,
holdings_scrip: parse_id_number_or_zero(&raw.holdings_scrip)?,
total_shares: parse_id_number(&raw.total_holding_shares)?,
percentage_bps: parse_id_percentage(&raw.percentage)?,
report_date: parse_ksei_date(&raw.date)?,
})
}
/// Find or create a canonical entity for a raw investor name.
/// Strategy: exact match on normalized name → rule-based normalization → create new.
/// Returns entity_id.
pub fn resolve_entity(
conn: &Connection,
raw_name: &str,
source: OwnershipSource,
) -> Result<i64, IdxError> {
let source_db = source_to_db(source);
let raw_trimmed = raw_name.trim();
if raw_trimmed.is_empty() {
return Err(IdxError::ParseError("empty raw entity name".to_string()));
}
let exact_normalized = collapse_whitespace(raw_trimmed).to_uppercase();
let rule_normalized = normalize_name(raw_trimmed);
if let Some(entity_id) = find_entity_by_alias(conn, raw_trimmed, source_db)? {
return Ok(entity_id);
}
if let Some(entity_id) = find_entity_by_canonical(conn, &exact_normalized)? {
insert_alias(conn, entity_id, raw_trimmed, source_db, "exact")?;
return Ok(entity_id);
}
if rule_normalized != exact_normalized
&& let Some(entity_id) = find_entity_by_canonical(conn, &rule_normalized)?
{
insert_alias(conn, entity_id, raw_trimmed, source_db, "rule")?;
return Ok(entity_id);
}
let now: i64 = 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![
if rule_normalized.is_empty() {
&exact_normalized
} else {
&rule_normalized
},
now
],
)
.map_err(|e| IdxError::DatabaseError(format!("insert entity failed: {e}")))?;
let entity_id = conn.last_insert_rowid();
insert_alias(conn, entity_id, raw_trimmed, source_db, "exact")?;
Ok(entity_id)
}
pub(crate) fn normalize_investor_type(raw: &str) -> Option<InvestorTypeCode> {
let value = raw.trim();
if value.is_empty() {
None
} else {
Some(InvestorTypeCode(value.to_uppercase()))
}
}
pub(crate) fn normalize_locality(raw: &str) -> Option<Locality> {
match raw.trim().to_uppercase().as_str() {
"L" | "D" => Some(Locality::Local),
"F" | "A" => Some(Locality::Foreign),
_ => None,
}
}
fn optional_string(raw: &str) -> Option<String> {
let value = raw.trim();
if value.is_empty() {
None
} else {
Some(value.to_string())
}
}
fn collapse_whitespace(input: &str) -> String {
input.split_whitespace().collect::<Vec<_>>().join(" ")
}
fn source_to_db(source: OwnershipSource) -> &'static str {
match source {
OwnershipSource::Ksei => "ksei",
OwnershipSource::Bing => "bing",
}
}
fn find_entity_by_alias(
conn: &Connection,
raw_name: &str,
source: &str,
) -> Result<Option<i64>, IdxError> {
conn.query_row(
"SELECT entity_id FROM entity_aliases WHERE raw_name = ?1 AND source = ?2",
params![raw_name, source],
|row| row.get(0),
)
.map(Some)
.or_else(|e| match e {
rusqlite::Error::QueryReturnedNoRows => Ok(None),
_ => Err(IdxError::DatabaseError(e.to_string())),
})
}
fn find_entity_by_canonical(
conn: &Connection,
canonical_name: &str,
) -> Result<Option<i64>, IdxError> {
conn.query_row(
"SELECT id FROM entities WHERE canonical_name = ?1",
params![canonical_name],
|row| row.get(0),
)
.map(Some)
.or_else(|e| match e {
rusqlite::Error::QueryReturnedNoRows => Ok(None),
_ => Err(IdxError::DatabaseError(e.to_string())),
})
}
fn insert_alias(
conn: &Connection,
entity_id: i64,
raw_name: &str,
source: &str,
method: &str,
) -> Result<(), IdxError> {
conn.execute(
"INSERT OR IGNORE INTO entity_aliases (entity_id, raw_name, source, confidence, method)
VALUES (?1, ?2, ?3, 1.0, ?4)",
params![entity_id, raw_name, source, method],
)
.map_err(|e| IdxError::DatabaseError(format!("insert alias failed: {e}")))?;
Ok(())
}
#[cfg(test)]
mod tests {
use chrono::NaiveDate;
use rusqlite::Connection;
use crate::ownership::entities::{
normalize_ksei_row, normalize_name, parse_id_number, parse_id_percentage, parse_ksei_date,
resolve_entity,
};
use crate::ownership::types::{KseiRawRow, OwnershipSource};
#[test]
fn test_parse_id_number() {
assert_eq!(parse_id_number("1.533.682.440").unwrap(), 1_533_682_440);
assert_eq!(parse_id_number("0").unwrap(), 0);
assert_eq!(parse_id_number("3.200.142.830").unwrap(), 3_200_142_830);
assert!(parse_id_number("").is_err());
}
#[test]
fn test_parse_id_percentage() {
assert_eq!(parse_id_percentage("54,94").unwrap(), 5494);
assert_eq!(parse_id_percentage("0,00").unwrap(), 0);
assert_eq!(parse_id_percentage("100,00").unwrap(), 10000);
assert_eq!(parse_id_percentage("41,10").unwrap(), 4110);
}
#[test]
fn test_parse_ksei_date_various_months() {
let samples = [
("01-Jan-2026", NaiveDate::from_ymd_opt(2026, 1, 1).unwrap()),
("01-Feb-2026", NaiveDate::from_ymd_opt(2026, 2, 1).unwrap()),
("01-Mar-2026", NaiveDate::from_ymd_opt(2026, 3, 1).unwrap()),
("01-Apr-2026", NaiveDate::from_ymd_opt(2026, 4, 1).unwrap()),
("01-May-2026", NaiveDate::from_ymd_opt(2026, 5, 1).unwrap()),
("01-Jun-2026", NaiveDate::from_ymd_opt(2026, 6, 1).unwrap()),
("01-Jul-2026", NaiveDate::from_ymd_opt(2026, 7, 1).unwrap()),
("01-Aug-2026", NaiveDate::from_ymd_opt(2026, 8, 1).unwrap()),
("01-Sep-2026", NaiveDate::from_ymd_opt(2026, 9, 1).unwrap()),
("01-Oct-2026", NaiveDate::from_ymd_opt(2026, 10, 1).unwrap()),
("01-Nov-2026", NaiveDate::from_ymd_opt(2026, 11, 1).unwrap()),
("01-Dec-2026", NaiveDate::from_ymd_opt(2026, 12, 1).unwrap()),
];
for (input, expected) in samples {
assert_eq!(parse_ksei_date(input).unwrap(), expected);
}
assert_eq!(
parse_ksei_date("27-Feb-2026").unwrap(),
NaiveDate::from_ymd_opt(2026, 2, 27).unwrap()
);
}
#[test]
fn test_normalize_name() {
assert_eq!(
normalize_name("PT. ASTRA INTERNATIONAL TBK"),
"ASTRA INTERNATIONAL"
);
assert_eq!(
normalize_name("PT BANK CENTRAL ASIA Tbk"),
"BANK CENTRAL ASIA"
);
assert_eq!(
normalize_name("UOB KAY HIAN PRIVATE LIMITED"),
"UOB KAY HIAN PRIVATE"
);
assert_eq!(
normalize_name("DJS Ketenagakerjaan (JHT)"),
"DJS KETENAGAKERJAAN (JHT)"
);
assert_eq!(
normalize_name("BPJS KETENAGAKERJAAN"),
"BPJS KETENAGAKERJAAN"
);
}
#[test]
fn test_resolve_entity_create_then_reuse() {
let conn = Connection::open_in_memory().unwrap();
conn.execute_batch(
r#"
CREATE TABLE 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 entity_aliases (
id INTEGER PRIMARY KEY,
entity_id INTEGER NOT NULL,
raw_name TEXT NOT NULL,
source TEXT NOT NULL,
confidence REAL NOT NULL DEFAULT 1.0,
method TEXT NOT NULL,
UNIQUE(raw_name, source)
);
"#,
)
.unwrap();
let id1 =
resolve_entity(&conn, "PT. ASTRA INTERNATIONAL TBK", OwnershipSource::Ksei).unwrap();
let id2 =
resolve_entity(&conn, "PT. ASTRA INTERNATIONAL TBK", OwnershipSource::Ksei).unwrap();
assert_eq!(id1, id2);
}
#[test]
fn test_normalize_ksei_row_defaults_empty_component_shares_to_zero() {
let raw = KseiRawRow {
date: "27-Apr-2026".to_string(),
share_code: "AADI".to_string(),
issuer_name: "ADARO ANDALAN INDONESIA Tbk".to_string(),
investor_name: "ADARO STRATEGIC INVESTMENTS".to_string(),
investor_type: "CP".to_string(),
local_foreign: "D".to_string(),
nationality: "INDONESIA".to_string(),
domicile: String::new(),
holdings_scripless: String::new(),
holdings_scrip: String::new(),
total_holding_shares: "3.200.142.830".to_string(),
percentage: "66,18".to_string(),
};
let draft = normalize_ksei_row(&raw).expect("empty component shares should normalize");
assert_eq!(draft.holdings_scripless, 0);
assert_eq!(draft.holdings_scrip, 0);
assert_eq!(draft.total_shares, 3_200_142_830);
}
#[test]
fn test_normalize_ksei_row_still_requires_total_shares() {
let raw = KseiRawRow {
date: "27-Apr-2026".to_string(),
share_code: "AADI".to_string(),
issuer_name: "ADARO ANDALAN INDONESIA Tbk".to_string(),
investor_name: "ADARO STRATEGIC INVESTMENTS".to_string(),
investor_type: "CP".to_string(),
local_foreign: "D".to_string(),
nationality: "INDONESIA".to_string(),
domicile: String::new(),
holdings_scripless: String::new(),
holdings_scrip: String::new(),
total_holding_shares: String::new(),
percentage: "66,18".to_string(),
};
assert!(normalize_ksei_row(&raw).is_err());
}
}

View file

@ -0,0 +1,462 @@
use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet};
use rusqlite::{Connection, params};
use crate::error::IdxError;
use crate::ownership::types::{GraphEdge, GraphNode, GraphNodeType, OwnershipSource};
/// Query ownership graph starting from a ticker or entity, traversing N hops.
/// Root can be ticker code (`BBCA`) or entity name — auto-detect.
pub fn query_ownership_graph(
conn: &Connection,
root: &str,
depth: usize,
) -> Result<(Vec<GraphNode>, Vec<GraphEdge>), IdxError> {
let root = root.trim();
if root.is_empty() {
return Err(IdxError::ParseError(
"graph root cannot be empty".to_string(),
));
}
let root_node_id = detect_root_node(conn, root)?;
let mut visited: BTreeSet<String> = BTreeSet::new();
{
let mut stmt = conn
.prepare(
"WITH RECURSIVE
latest_ksei AS (
SELECT sha256
FROM ownership_releases
ORDER BY as_of_date DESC, imported_at DESC
LIMIT 1
),
latest_bing AS (
SELECT ticker_id, MAX(report_date) AS report_date
FROM bing_holdings
GROUP BY ticker_id
),
all_edges AS (
SELECT
'entity:' || k.entity_id AS from_id,
'ticker:' || t.code AS to_id
FROM ksei_holdings k
JOIN tickers t ON t.id = k.ticker_id
JOIN latest_ksei lk ON lk.sha256 = k.release_sha256
WHERE k.entity_id IS NOT NULL
UNION
SELECT
'entity:' || b.entity_id AS from_id,
'ticker:' || t.code AS to_id
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 IS NOT NULL
AND b.signal = 'holder'
),
neighbors AS (
SELECT from_id AS a, to_id AS b FROM all_edges
UNION
SELECT to_id AS a, from_id AS b FROM all_edges
),
walk(node_id, depth, path) AS (
SELECT ?1 AS node_id, 0 AS depth, ?1 AS path
UNION ALL
SELECT n.b, w.depth + 1, w.path || '>' || n.b
FROM walk w
JOIN neighbors n ON n.a = w.node_id
WHERE w.depth < ?2
AND instr(w.path, n.b) = 0
)
SELECT DISTINCT node_id
FROM walk",
)
.map_err(|e| IdxError::DatabaseError(e.to_string()))?;
let rows = stmt
.query_map(params![root_node_id, depth as i64], |row| {
row.get::<_, String>(0)
})
.map_err(|e| IdxError::DatabaseError(e.to_string()))?;
for row in rows {
visited.insert(row.map_err(|e| IdxError::DatabaseError(e.to_string()))?);
}
}
let mut edges = query_all_edges(conn)?;
edges.retain(|edge| visited.contains(&edge.from) && visited.contains(&edge.to));
let nodes = build_nodes(conn, &visited)?;
Ok((nodes, edges))
}
/// Format graph as ASCII tree for terminal display.
pub fn format_graph_text(nodes: &[GraphNode], edges: &[GraphEdge]) -> String {
if nodes.is_empty() {
return "(empty graph)".to_string();
}
let labels: HashMap<&str, (&str, GraphNodeType)> = nodes
.iter()
.map(|n| (n.id.as_str(), (n.label.as_str(), n.node_type)))
.collect();
let mut ticker_to_entities: BTreeMap<&str, Vec<&GraphEdge>> = BTreeMap::new();
for edge in edges {
if let Some((_, GraphNodeType::Ticker)) = labels.get(edge.to.as_str()) {
ticker_to_entities
.entry(edge.to.as_str())
.or_default()
.push(edge);
}
}
let mut out = String::new();
out.push_str(&format!("nodes: {} edges: {}\n", nodes.len(), edges.len()));
for (ticker_id, mut rels) in ticker_to_entities {
rels.sort_by_key(|edge| std::cmp::Reverse(edge.percentage_bps));
let ticker_label = labels.get(ticker_id).map(|v| v.0).unwrap_or(ticker_id);
out.push_str(&format!("\n{ticker_label} [TICKER]\n"));
for (idx, edge) in rels.iter().enumerate() {
let holder_label = labels.get(edge.from.as_str()).map(|v| v.0).unwrap_or("?");
let branch = if idx + 1 == rels.len() {
"└─"
} else {
"├─"
};
out.push_str(&format!(
"{branch} {holder_label} ({:.2}%, {})\n",
edge.percentage_bps as f64 / 100.0,
source_label(edge.source)
));
}
}
if edges.is_empty() {
out.push_str("\n(no ownership edges found)\n");
}
out
}
/// Format graph as Graphviz DOT for export.
pub fn format_graph_dot(nodes: &[GraphNode], edges: &[GraphEdge]) -> String {
let mut out = String::new();
out.push_str("digraph ownership {\n");
out.push_str(" rankdir=LR;\n");
out.push_str(" graph [fontname=\"Helvetica\"];\n");
out.push_str(" node [fontname=\"Helvetica\"];\n");
out.push_str(" edge [fontname=\"Helvetica\"];\n\n");
for node in nodes {
let shape = match node.node_type {
GraphNodeType::Entity => "ellipse",
GraphNodeType::Ticker => "box",
};
out.push_str(&format!(
" \"{}\" [label=\"{}\", shape={}];\n",
escape_dot(&node.id),
escape_dot(&node.label),
shape
));
}
out.push('\n');
for edge in edges {
out.push_str(&format!(
" \"{}\" -> \"{}\" [label=\"{:.2}% ({})\"];\n",
escape_dot(&edge.from),
escape_dot(&edge.to),
edge.percentage_bps as f64 / 100.0,
source_label(edge.source)
));
}
out.push_str("}\n");
out
}
fn detect_root_node(conn: &Connection, root: &str) -> Result<String, IdxError> {
let maybe_ticker = conn
.query_row(
"SELECT code FROM tickers WHERE UPPER(code) = UPPER(?1) LIMIT 1",
params![root],
|row| row.get::<_, String>(0),
)
.ok();
if let Some(code) = maybe_ticker {
return Ok(format!("ticker:{code}"));
}
let q = format!("%{root}%");
let maybe_entity_id = conn
.query_row(
"SELECT id
FROM entities
WHERE canonical_name LIKE ?1 COLLATE NOCASE
ORDER BY LENGTH(canonical_name) ASC, canonical_name ASC
LIMIT 1",
params![q],
|row| row.get::<_, i64>(0),
)
.ok();
if let Some(entity_id) = maybe_entity_id {
return Ok(format!("entity:{entity_id}"));
}
Err(IdxError::ParseError(format!(
"graph root not found as ticker or entity: {root}"
)))
}
fn query_all_edges(conn: &Connection) -> Result<Vec<GraphEdge>, IdxError> {
let mut out = Vec::new();
if let Ok(release_sha) = conn.query_row(
"SELECT sha256
FROM ownership_releases
ORDER BY as_of_date DESC, imported_at DESC
LIMIT 1",
[],
|row| row.get::<_, String>(0),
) {
let mut stmt = conn
.prepare(
"SELECT 'entity:' || k.entity_id,
'ticker:' || t.code,
k.percentage_bps
FROM ksei_holdings k
JOIN tickers t ON t.id = k.ticker_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![release_sha], |row| {
Ok(GraphEdge {
from: row.get(0)?,
to: row.get(1)?,
percentage_bps: row.get(2)?,
source: OwnershipSource::Ksei,
})
})
.map_err(|e| IdxError::DatabaseError(e.to_string()))?;
for row in rows {
out.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 'entity:' || b.entity_id,
'ticker:' || t.code,
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 tickers t ON t.id = b.ticker_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(GraphEdge {
from: row.get(0)?,
to: row.get(1)?,
percentage_bps: row.get(2)?,
source: OwnershipSource::Bing,
})
})
.map_err(|e| IdxError::DatabaseError(e.to_string()))?;
for row in rows {
out.push(row.map_err(|e| IdxError::DatabaseError(e.to_string()))?);
}
}
dedup_edges(out)
}
fn dedup_edges(edges: Vec<GraphEdge>) -> Result<Vec<GraphEdge>, IdxError> {
let mut seen: HashSet<(String, String, i64, &'static str)> = HashSet::new();
let mut out = Vec::new();
for edge in edges {
let key = (
edge.from.clone(),
edge.to.clone(),
edge.percentage_bps,
source_label(edge.source),
);
if seen.insert(key) {
out.push(edge);
}
}
Ok(out)
}
fn build_nodes(conn: &Connection, node_ids: &BTreeSet<String>) -> Result<Vec<GraphNode>, IdxError> {
let mut nodes = Vec::new();
for node_id in node_ids {
if let Some(entity_id) = node_id.strip_prefix("entity:") {
let entity_id_num = entity_id.parse::<i64>().map_err(|e| {
IdxError::ParseError(format!("invalid entity node id '{node_id}': {e}"))
})?;
let label = conn
.query_row(
"SELECT canonical_name FROM entities WHERE id = ?1",
params![entity_id_num],
|row| row.get::<_, String>(0),
)
.unwrap_or_else(|_| format!("entity:{entity_id_num}"));
nodes.push(GraphNode {
id: node_id.clone(),
label,
node_type: GraphNodeType::Entity,
});
continue;
}
if let Some(code) = node_id.strip_prefix("ticker:") {
nodes.push(GraphNode {
id: node_id.clone(),
label: code.to_string(),
node_type: GraphNodeType::Ticker,
});
}
}
Ok(nodes)
}
fn source_label(source: OwnershipSource) -> &'static str {
match source {
OwnershipSource::Ksei => "ksei",
OwnershipSource::Bing => "bing",
}
}
fn escape_dot(value: &str) -> String {
value.replace('"', "\\\"")
}
#[cfg(test)]
mod tests {
use chrono::NaiveDate;
use rusqlite::Connection;
use crate::ownership::db::{ensure_schema, upsert_ticker, write_ksei_release};
use crate::ownership::types::{KseiHolding, OwnershipRelease};
use super::query_ownership_graph;
fn setup() -> Connection {
let conn = Connection::open_in_memory().unwrap();
ensure_schema(&conn).unwrap();
conn
}
#[test]
fn graph_uses_latest_ksei_release_only() {
let conn = setup();
let aa = upsert_ticker(&conn, "AA", None).unwrap();
conn.execute(
"INSERT INTO entities (canonical_name, entity_type, country, created_at, updated_at)
VALUES ('OLDER HOLDER', NULL, NULL, 0, 0)",
[],
)
.unwrap();
let older_entity = conn.last_insert_rowid();
conn.execute(
"INSERT INTO entities (canonical_name, entity_type, country, created_at, updated_at)
VALUES ('LATEST HOLDER', NULL, NULL, 0, 0)",
[],
)
.unwrap();
let latest_entity = conn.last_insert_rowid();
let old_release = OwnershipRelease {
id: 0,
source_url: None,
sha256: "old-graph".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: Some(older_entity),
raw_investor_name: "OLDER HOLDER".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(),
}];
write_ksei_release(&conn, &old_release, &old_rows, false).unwrap();
let latest_release = OwnershipRelease {
id: 0,
source_url: None,
sha256: "latest-graph".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(latest_entity),
raw_investor_name: "LATEST HOLDER".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: latest_release.as_of_date,
release_sha256: latest_release.sha256.clone(),
}];
write_ksei_release(&conn, &latest_release, &latest_rows, false).unwrap();
let (nodes, edges) = query_ownership_graph(&conn, "AA", 1).unwrap();
assert!(nodes.iter().any(|node| node.label == "LATEST HOLDER"));
assert!(!nodes.iter().any(|node| node.label == "OLDER HOLDER"));
assert_eq!(edges.len(), 1);
assert_eq!(edges[0].percentage_bps, 1200);
}
}

View file

@ -0,0 +1,10 @@
pub mod archive;
pub mod db;
pub mod entities;
pub mod graph;
pub mod parser;
pub mod remote;
pub mod search;
pub mod snapshot;
pub mod types;
pub mod xlsx;

View file

@ -0,0 +1,862 @@
use std::collections::HashMap;
use std::path::Path;
use std::process::Command;
use quick_xml::Reader;
use quick_xml::events::{BytesStart, Event};
use crate::error::IdxError;
use crate::ownership::types::KseiRawRow;
const Y_TOLERANCE: f32 = 0.8;
const HEADER_MATCH_MIN: usize = 4;
const HEADER_LABELS: &[&str] = &[
"DATE",
"SHARECODE",
"ISSUERNAME",
"INVESTORNAME",
"INVESTORTYPE",
"INVESTORCLASSIFICATION",
"LOCALFOREIGN",
"NATIONALITY",
"DOMICILE",
"HOLDINGSSCRIPLESS",
"HOLDINGSSCRIP",
"TOTALHOLDINGSHARES",
"PERCENTAGE",
];
const HOLDER_REGISTER_SCHEMA_MARKERS: &[&str] = &[
"TEXT=\"DATE\"",
"TEXT=\"SHARE_CODE\"",
"TEXT=\"INVESTOR_NAME\"",
"TEXT=\"INVESTOR_TYPE\"",
"TEXT=\"INVESTOR_CLASSIFICATION\"",
"TEXT=\"LOCAL_FOREIGN\"",
"TEXT=\"TOTAL_HOLDING_SHARES\"",
"TEXT=\"PERCENTAGE\"",
];
const ANNOUNCEMENT_WRAPPER_SCHEMA_MARKERS: &[&str] =
&["TEXT=\"PENGUMUMAN\"", "PT_BURSA_EFEK_INDONESIA_(BEI)"];
const ABOVE_FIVE_SCHEMA_MARKERS: &[&str] =
&["TEXT=\"INVS\"", "REKENING_TAMPUNGAN_KSEI", "CLOSED_MEMBER-"];
const INVESTOR_TYPE_SCHEMA_MARKERS: &[&str] = &[
"TEXT=\"STOCK_CODE\"",
"TEXT=\"NUMBER_OF_SHARES\"",
"TEXT=\"FOREIGN\"",
];
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum OwnershipPdfSchema {
HolderRegister,
AnnouncementWrapper,
LegacyAboveFivePercent,
LegacyInvestorType,
Unknown,
}
impl OwnershipPdfSchema {
pub fn label(self) -> &'static str {
match self {
Self::HolderRegister => "holder_register",
Self::AnnouncementWrapper => "announcement_wrapper",
Self::LegacyAboveFivePercent => "legacy_above5",
Self::LegacyInvestorType => "legacy_investor_type",
Self::Unknown => "unknown",
}
}
}
#[derive(Debug, Clone)]
struct PageLine {
x: f32,
y: f32,
text: String,
}
/// Extract mutool stext XML from a PDF file.
pub fn extract_pdf_stext(path: &Path) -> Result<String, IdxError> {
check_mutool()?;
let output = Command::new("mutool")
.arg("convert")
.arg("-F")
.arg("stext")
.arg("-o")
.arg("-")
.arg(path)
.output()
.map_err(|e| IdxError::PdfParseError(format!("failed to run mutool: {e}")))?;
if !output.status.success() {
let stderr = String::from_utf8_lossy(&output.stderr);
return Err(IdxError::PdfParseError(format!(
"mutool convert failed: {}",
stderr.trim()
)));
}
let xml = String::from_utf8(output.stdout)
.map_err(|e| IdxError::PdfParseError(format!("invalid utf-8 stext output: {e}")))?;
Ok(xml)
}
/// Classify a PDF schema from mutool stext XML before the row parser runs.
pub fn classify_stext_xml(xml: &str) -> OwnershipPdfSchema {
let normalized = normalize_stext_for_classification(xml);
if count_schema_markers(&normalized, HOLDER_REGISTER_SCHEMA_MARKERS) >= 5 {
return OwnershipPdfSchema::HolderRegister;
}
if count_schema_markers(&normalized, ABOVE_FIVE_SCHEMA_MARKERS) >= 2 {
return OwnershipPdfSchema::LegacyAboveFivePercent;
}
if count_schema_markers(&normalized, INVESTOR_TYPE_SCHEMA_MARKERS) >= 3 {
return OwnershipPdfSchema::LegacyInvestorType;
}
if count_schema_markers(&normalized, ANNOUNCEMENT_WRAPPER_SCHEMA_MARKERS) >= 2 {
return OwnershipPdfSchema::AnnouncementWrapper;
}
OwnershipPdfSchema::Unknown
}
/// Normalize stext XML for schema classification.
///
/// Newer KSEI PDFs use spaces in header text (e.g. `SHARE CODE` instead of
/// `SHARE_CODE`, `INVESTOR CLASSIFICATION` instead of `INVESTOR_TYPE`). The
/// schema markers use the underscored form, so we replace spaces with
/// underscores inside `text="..."` attribute values to match both layouts.
fn normalize_stext_for_classification(xml: &str) -> String {
let upper = xml.to_ascii_uppercase();
let mut result = String::with_capacity(upper.len());
let bytes = upper.as_bytes();
let mut i = 0;
while i < bytes.len() {
if bytes[i..].starts_with(b"TEXT=\"") {
result.push_str("TEXT=\"");
i += 6; // skip past TEXT="
// Collect everything until the closing quote.
while i < bytes.len() && bytes[i] != b'"' {
let ch = bytes[i] as char;
result.push(if ch == ' ' { '_' } else { ch });
i += 1;
}
if i < bytes.len() {
result.push('"');
i += 1; // skip closing quote
}
} else {
result.push(bytes[i] as char);
i += 1;
}
}
result
}
/// Parse a KSEI ownership PDF into raw rows.
/// Shells out to `mutool` for XML extraction, classifies the schema,
/// and only parses the supported holder-register layout.
pub fn parse_ksei_pdf(path: &Path) -> Result<Vec<KseiRawRow>, IdxError> {
let xml = extract_pdf_stext(path)?;
ensure_supported_schema(classify_stext_xml(&xml))?;
parse_stext_xml(&xml)
}
/// Parse mutool stext XML output into raw rows.
/// Pure function — takes XML string, returns parsed rows.
pub fn parse_stext_xml(xml: &str) -> Result<Vec<KseiRawRow>, IdxError> {
let mut reader = Reader::from_str(xml);
reader.config_mut().trim_text(false);
let mut rows: Vec<KseiRawRow> = Vec::new();
let mut current_page: Option<Vec<PageLine>> = None;
let mut buf = Vec::new();
loop {
match reader.read_event_into(&mut buf) {
Ok(Event::Start(e)) if e.name().as_ref() == b"page" => {
current_page = Some(Vec::new());
}
Ok(Event::Start(e)) if e.name().as_ref() == b"line" => {
if let Some(page) = current_page.as_mut()
&& let Some(line) = parse_line_attrs(&reader, &e)?
{
page.push(line);
}
}
Ok(Event::End(e)) if e.name().as_ref() == b"page" => {
if let Some(page) = current_page.take() {
rows.extend(extract_rows_from_page(&page));
}
}
Ok(Event::Eof) => break,
Err(err) => {
return Err(IdxError::PdfParseError(format!(
"failed to parse stext XML: {err}"
)));
}
_ => {}
}
buf.clear();
}
Ok(rows)
}
/// Check if mutool is available in PATH.
pub fn check_mutool() -> Result<(), IdxError> {
// mutool with no args prints usage to stderr and exits non-zero,
// so we just check that the binary is found and executable.
Command::new("mutool")
.arg("--help")
.stdout(std::process::Stdio::null())
.stderr(std::process::Stdio::null())
.status()
.map_err(|e| IdxError::PdfParseError(format!("mutool not found in PATH: {e}")))?;
Ok(())
}
fn ensure_supported_schema(schema: OwnershipPdfSchema) -> Result<(), IdxError> {
match schema {
OwnershipPdfSchema::HolderRegister => Ok(()),
OwnershipPdfSchema::AnnouncementWrapper => Err(IdxError::Unsupported(
"IDX announcement wrapper PDFs are not importable; run `idx ownership discover` and use the `lamp1` attachment URL".to_string(),
)),
OwnershipPdfSchema::LegacyAboveFivePercent => Err(IdxError::Unsupported(
"legacy IDX `above5` ownership PDFs are not supported for import; only the `above1` holder-register `lamp1` attachment is supported".to_string(),
)),
OwnershipPdfSchema::LegacyInvestorType => Err(IdxError::Unsupported(
"legacy IDX `investor-type` ownership PDFs are not supported for import; only the `above1` holder-register `lamp1` attachment is supported".to_string(),
)),
OwnershipPdfSchema::Unknown => Err(IdxError::ParseError(
"PDF did not match the supported KSEI holder-register layout".to_string(),
)),
}
}
fn count_schema_markers(haystack: &str, markers: &[&str]) -> usize {
markers
.iter()
.filter(|marker| haystack.contains(**marker))
.count()
}
fn parse_line_attrs(
reader: &Reader<&[u8]>,
event: &BytesStart<'_>,
) -> Result<Option<PageLine>, IdxError> {
let mut bbox: Option<String> = None;
let mut text: Option<String> = None;
for attr_result in event.attributes().with_checks(false) {
let attr = attr_result
.map_err(|err| IdxError::PdfParseError(format!("invalid XML attribute: {err}")))?;
match attr.key.as_ref() {
b"bbox" => {
bbox = Some(
attr.decode_and_unescape_value(reader.decoder())
.map_err(|err| {
IdxError::PdfParseError(format!("invalid XML bbox attribute: {err}"))
})?
.to_string(),
);
}
b"text" => {
text = Some(
attr.decode_and_unescape_value(reader.decoder())
.map_err(|err| {
IdxError::PdfParseError(format!("invalid XML text attribute: {err}"))
})?
.to_string(),
);
}
_ => {}
}
}
let Some(text) = text.map(|value| normalize_spaces(&value)) else {
return Ok(None);
};
if text.is_empty() {
return Ok(None);
}
let Some(bbox) = bbox else {
return Ok(None);
};
let mut parts = bbox.split_whitespace();
let x = parts.next().and_then(|value| value.parse::<f32>().ok());
let y = parts.next().and_then(|value| value.parse::<f32>().ok());
match (x, y) {
(Some(x), Some(y)) => Ok(Some(PageLine { x, y, text })),
_ => Ok(None),
}
}
fn extract_rows_from_page(page: &[PageLine]) -> Vec<KseiRawRow> {
let mut rows_by_y: HashMap<i32, Vec<PageLine>> = HashMap::new();
for line in page {
rows_by_y
.entry(y_bucket(line.y))
.or_default()
.push(line.clone());
}
let mut y_keys: Vec<i32> = rows_by_y.keys().copied().collect();
y_keys.sort_unstable();
let mut rows = Vec::new();
for y in y_keys {
let Some(lines) = rows_by_y.get(&y) else {
continue;
};
let mut sorted = lines.clone();
sorted.sort_by(|left, right| left.x.total_cmp(&right.x));
let texts: Vec<String> = sorted.into_iter().map(|line| line.text).collect();
if is_header_row(&texts) {
continue;
}
let row = parse_row_segments(&texts);
if is_data_row(&row) {
rows.push(row);
}
}
rows
}
fn is_header_row(texts: &[String]) -> bool {
texts
.iter()
.filter(|text| HEADER_LABELS.contains(&normalize_header_label(text).as_str()))
.count()
>= HEADER_MATCH_MIN
}
fn parse_row_segments(texts: &[String]) -> KseiRawRow {
let mut row = KseiRawRow {
date: String::new(),
share_code: String::new(),
issuer_name: String::new(),
investor_name: String::new(),
investor_type: String::new(),
local_foreign: String::new(),
nationality: String::new(),
domicile: String::new(),
holdings_scripless: String::new(),
holdings_scrip: String::new(),
total_holding_shares: String::new(),
percentage: String::new(),
};
let mut remaining: Vec<String> = texts
.iter()
.map(|text| normalize_spaces(text))
.filter(|text| !text.is_empty())
.collect();
if remaining.is_empty() {
return row;
}
if let Some((date, share_code)) = split_date_and_share(&remaining[0]) {
row.date = date;
row.share_code = share_code;
let _ = remaining.remove(0);
}
if row.date.is_empty() {
return row;
}
if row.share_code.is_empty()
&& remaining
.first()
.is_some_and(|segment| is_share_code_like(segment))
{
row.share_code = remaining.remove(0);
}
if remaining
.last()
.is_some_and(|segment| is_percentage_like(segment))
{
row.percentage = remaining.pop().unwrap_or_default();
}
let numeric_tail = pop_numeric_tail(&mut remaining);
match numeric_tail.as_slice() {
[scripless, scrip, total] => {
row.holdings_scripless = scripless.clone();
row.holdings_scrip = scrip.clone();
row.total_holding_shares = total.clone();
}
// Some PDFs omit the zero-valued scrip column entirely.
[scripless, total] => {
row.holdings_scripless = scripless.clone();
row.holdings_scrip = "0".to_string();
row.total_holding_shares = total.clone();
}
[total] => {
row.total_holding_shares = total.clone();
}
_ => {}
}
let mut geo_fields = Vec::new();
while remaining.len() > 2 {
let Some(candidate) = remaining.last().cloned() else {
break;
};
if row.local_foreign.is_empty() && is_locality_marker(&candidate) {
row.local_foreign = remaining.pop().unwrap_or_default();
continue;
}
if row.investor_type.is_empty() && is_investor_type_marker(&candidate) {
row.investor_type = remaining.pop().unwrap_or_default();
continue;
}
if geo_fields.len() < 2 {
geo_fields.push(remaining.pop().unwrap_or_default());
continue;
}
break;
}
geo_fields.reverse();
if let Some(first) = geo_fields.first() {
row.nationality = first.clone();
}
if geo_fields.len() > 1 {
row.domicile = geo_fields[1..].join(" ");
}
if let Some(first) = remaining.first() {
row.issuer_name = first.clone();
}
if remaining.len() > 1 {
row.investor_name = remaining[1..].join(" ");
}
row
}
fn split_date_and_share(segment: &str) -> Option<(String, String)> {
let trimmed = segment.trim();
if trimmed.len() < 11 {
return None;
}
let date = trimmed.get(..11)?.to_string();
if !is_ksei_date(&date) {
return None;
}
let share_code = trimmed.get(11..).unwrap_or_default().trim().to_string();
Some((date, share_code))
}
fn is_share_code_like(value: &str) -> bool {
let trimmed = value.trim();
(3..=8).contains(&trimmed.len())
&& trimmed
.chars()
.all(|ch| ch.is_ascii_uppercase() || ch.is_ascii_digit())
}
fn pop_numeric_tail(remaining: &mut Vec<String>) -> Vec<String> {
let mut numeric_tail = Vec::new();
while numeric_tail.len() < 3
&& remaining
.last()
.is_some_and(|segment| is_id_number_like(segment))
{
numeric_tail.push(remaining.pop().unwrap_or_default());
}
numeric_tail.reverse();
numeric_tail
}
fn is_id_number_like(value: &str) -> bool {
let trimmed = value.trim();
!trimmed.is_empty()
&& trimmed != "-"
&& trimmed.chars().all(|ch| ch.is_ascii_digit() || ch == '.')
}
fn is_investor_type_marker(value: &str) -> bool {
let trimmed = value.trim();
!trimmed.is_empty() && trimmed.len() <= 4 && trimmed.chars().all(|ch| ch.is_ascii_uppercase())
}
fn is_locality_marker(value: &str) -> bool {
matches!(
value.trim().to_ascii_uppercase().as_str(),
"L" | "F" | "A" | "D" | "LOCAL" | "FOREIGN" | "ASING" | "DOMESTIC"
)
}
fn normalize_header_label(text: &str) -> String {
let mut normalized = String::new();
for ch in text.chars() {
if ch.is_ascii_alphanumeric() {
normalized.push(ch.to_ascii_uppercase());
}
}
normalized
}
fn y_bucket(y: f32) -> i32 {
(y / Y_TOLERANCE).round() as i32
}
fn normalize_spaces(input: &str) -> String {
let mut out = String::with_capacity(input.len());
let mut prev_space = false;
for ch in input.chars() {
if ch.is_whitespace() {
if !prev_space {
out.push(' ');
}
prev_space = true;
} else {
out.push(ch);
prev_space = false;
}
}
out.trim().to_string()
}
fn is_data_row(row: &KseiRawRow) -> bool {
is_ksei_date(&row.date)
&& is_percentage_like(&row.percentage)
&& !row.share_code.trim().is_empty()
&& !row.investor_name.trim().is_empty()
}
fn is_ksei_date(s: &str) -> bool {
if s.len() != 11 {
return false;
}
let mut parts = s.split('-');
let day = parts.next();
let mon = parts.next();
let year = parts.next();
if parts.next().is_some() {
return false;
}
match (day, mon, year) {
(Some(d), Some(m), Some(y)) => {
d.len() == 2
&& d.chars().all(|c| c.is_ascii_digit())
&& m.len() == 3
&& m.chars().all(|c| c.is_ascii_alphabetic())
&& y.len() == 4
&& y.chars().all(|c| c.is_ascii_digit())
}
_ => false,
}
}
fn is_percentage_like(s: &str) -> bool {
let cleaned = s.trim();
if cleaned.is_empty() {
return false;
}
let mut parts = cleaned.split(',');
let left = parts.next();
let right = parts.next();
if parts.next().is_some() {
return false;
}
match (left, right) {
(Some(l), Some(r)) => {
!l.is_empty()
&& l.chars().all(|c| c.is_ascii_digit())
&& !r.is_empty()
&& r.chars().all(|c| c.is_ascii_digit())
}
_ => false,
}
}
#[cfg(test)]
mod tests {
use std::path::Path;
use crate::ownership::entities::normalize_ksei_row;
use super::{
OwnershipPdfSchema, check_mutool, classify_stext_xml, parse_ksei_pdf, parse_row_segments,
parse_stext_xml,
};
#[test]
fn test_parse_stext_xml_live_like_lines_extract_rows() {
let xml = build_live_like_stext_xml();
let rows = parse_stext_xml(&xml).expect("failed to parse live-like fixture XML");
assert_eq!(rows.len(), 3);
let first = &rows[0];
assert_eq!(first.date, "27-Feb-2026");
assert_eq!(first.share_code, "AADI");
assert_eq!(first.issuer_name, "ADARO ANDALAN INDONESIA Tbk");
assert_eq!(first.investor_name, "ADARO STRATEGIC INVESTMENTS");
assert_eq!(first.investor_type, "CP");
assert_eq!(first.local_foreign, "D");
assert_eq!(first.nationality, "INDONESIA");
assert_eq!(first.holdings_scripless, "3.200.142.830");
assert_eq!(first.holdings_scrip, "0");
assert_eq!(first.total_holding_shares, "3.200.142.830");
assert_eq!(first.percentage, "66,18");
let last = &rows[2];
assert_eq!(last.share_code, "BBRI");
assert_eq!(last.investor_name, "PT NUSANTARA CAPITAL");
assert_eq!(last.investor_type, "ID");
assert_eq!(last.local_foreign, "A");
assert_eq!(last.nationality, "SINGAPORE");
assert_eq!(last.percentage, "15,00");
}
#[test]
fn test_parse_stext_xml_real_above1_excerpt_extracts_row() {
let xml = include_str!("../../tests/fixtures/ksei_above1_stext_excerpt.xml");
let rows = parse_stext_xml(xml).expect("failed to parse real above1 excerpt XML");
assert_eq!(rows.len(), 1);
let row = &rows[0];
assert_eq!(row.date, "27-Feb-2026");
assert_eq!(row.share_code, "AADI");
assert_eq!(row.issuer_name, "ADARO ANDALAN INDONESIA Tbk");
assert_eq!(row.investor_name, "ADARO STRATEGIC INVESTMENTS");
assert_eq!(row.investor_type, "CP");
assert_eq!(row.local_foreign, "D");
assert_eq!(row.nationality, "INDONESIA");
assert_eq!(row.holdings_scripless, "3.200.142.830");
assert_eq!(row.holdings_scrip, "0");
assert_eq!(row.total_holding_shares, "3.200.142.830");
assert_eq!(row.percentage, "41,10");
}
#[test]
fn test_parse_row_segments_missing_scrip_defaults_to_zero() {
let texts = vec![
"27-Apr-2026 AADI".to_string(),
"ADARO ANDALAN INDONESIA Tbk".to_string(),
"ADARO STRATEGIC INVESTMENTS".to_string(),
"CP".to_string(),
"D".to_string(),
"INDONESIA".to_string(),
"3.200.142.830".to_string(),
"3.200.142.830".to_string(),
"66,18".to_string(),
];
let row = parse_row_segments(&texts);
assert_eq!(row.date, "27-Apr-2026");
assert_eq!(row.share_code, "AADI");
assert_eq!(row.holdings_scripless, "3.200.142.830");
assert_eq!(row.holdings_scrip, "0");
assert_eq!(row.total_holding_shares, "3.200.142.830");
assert_eq!(row.percentage, "66,18");
}
#[test]
fn test_parse_row_segments_missing_scrip_normalizes_without_error() {
let texts = vec![
"27-Apr-2026 AADI".to_string(),
"ADARO ANDALAN INDONESIA Tbk".to_string(),
"ADARO STRATEGIC INVESTMENTS".to_string(),
"CP".to_string(),
"D".to_string(),
"INDONESIA".to_string(),
"3.200.142.830".to_string(),
"3.200.142.830".to_string(),
"66,18".to_string(),
];
let row = parse_row_segments(&texts);
let draft = normalize_ksei_row(&row).expect("missing scrip column should normalize");
assert_eq!(draft.holdings_scripless, 3_200_142_830);
assert_eq!(draft.holdings_scrip, 0);
assert_eq!(draft.total_shares, 3_200_142_830);
}
#[test]
fn classify_stext_xml_detects_supported_holder_register_schema() {
let xml = include_str!("../../tests/fixtures/ksei_above1_stext_excerpt.xml");
assert_eq!(classify_stext_xml(xml), OwnershipPdfSchema::HolderRegister);
}
#[test]
fn classify_stext_xml_detects_announcement_wrapper_schema() {
let xml = include_str!("../../tests/fixtures/ksei_announcement_wrapper_stext_excerpt.xml");
assert_eq!(
classify_stext_xml(xml),
OwnershipPdfSchema::AnnouncementWrapper
);
}
#[test]
fn classify_stext_xml_detects_spaced_holder_register_schema() {
let xml = include_str!("../../tests/fixtures/ksei_above1_spaced_stext_excerpt.xml");
assert_eq!(classify_stext_xml(xml), OwnershipPdfSchema::HolderRegister);
}
#[test]
fn classify_stext_xml_detects_legacy_above5_schema() {
let xml = include_str!("../../tests/fixtures/ksei_above5_stext_excerpt.xml");
assert_eq!(
classify_stext_xml(xml),
OwnershipPdfSchema::LegacyAboveFivePercent
);
}
#[test]
fn classify_stext_xml_detects_legacy_investor_type_schema() {
let xml = include_str!("../../tests/fixtures/ksei_investor_type_stext_excerpt.xml");
assert_eq!(
classify_stext_xml(xml),
OwnershipPdfSchema::LegacyInvestorType
);
}
#[test]
fn test_parse_ksei_pdf_real_file_row_count() {
if check_mutool().is_err() {
eprintln!("skipping mutool-dependent test: mutool not available");
return;
}
let pdf_path =
Path::new("/var/lib/openclaw/projects/idx-cli/research/ownership_202603.pdf");
if !pdf_path.exists() {
eprintln!("skipping mutool-dependent test: sample PDF not found");
return;
}
let rows = parse_ksei_pdf(pdf_path).expect("failed to parse real KSEI PDF");
assert!(
rows.len() >= 7_200,
"expected at least 7200 rows, got {}",
rows.len()
);
}
fn build_live_like_stext_xml() -> String {
let mut xml = String::from(r#"<?xml version="1.0"?><document>"#);
append_page(
&mut xml,
"page1",
&[
(31.56, 86.33, "DATE"),
(56.64, 86.33, "SHARE_CODE"),
(121.22, 86.33, "ISSUER_NAME"),
(267.17, 86.33, "INVESTOR_NAME"),
(390.89, 86.33, "INVESTOR_TYPE"),
(434.11, 86.33, "LOCAL_FOREIGN"),
(475.87, 86.33, "NATIONALITY"),
(525.19, 86.33, "DOMICILE"),
(574.63, 86.33, "HOLDINGS_SCRIPLESS"),
(629.98, 86.33, "HOLDINGS_SCRIP"),
(680.02, 86.33, "TOTAL_HOLDING_SHARES"),
(741.22, 86.33, "PERCENTAGE"),
(28.68, 91.01, "27-Feb-2026 AADI"),
(85.10, 91.01, "ADARO ANDALAN INDONESIA Tbk"),
(179.30, 91.01, "ADARO STRATEGIC INVESTMENTS"),
(381.41, 91.01, "CP"),
(424.75, 91.01, "D"),
(504.07, 91.01, "INDONESIA"),
(597.58, 91.01, "3.200.142.830"),
(630.10, 91.01, "0"),
(680.12, 91.01, "3.200.142.830"),
(741.30, 91.01, "66,18"),
(28.68, 96.01, "27-Feb-2026 AADI"),
(85.10, 96.01, "ADARO ANDALAN INDONESIA Tbk"),
(179.30, 96.01, "PUBLIC"),
(381.41, 96.01, "OT"),
(424.75, 96.01, "A"),
(504.07, 96.01, "SINGAPORE"),
(597.58, 96.01, "500.000.000"),
(630.10, 96.01, "0"),
(680.12, 96.01, "500.000.000"),
(741.30, 96.01, "10,34"),
],
);
append_page(
&mut xml,
"page2",
&[
(28.68, 20.00, "27-Feb-2026 BBRI"),
(85.10, 20.00, "BANK RAKYAT INDONESIA Tbk"),
(179.30, 20.00, "PT NUSANTARA CAPITAL"),
(381.41, 20.00, "ID"),
(424.75, 20.00, "A"),
(504.07, 20.00, "SINGAPORE"),
(597.58, 20.00, "1.250.000.000"),
(630.10, 20.00, "0"),
(680.12, 20.00, "1.250.000.000"),
(741.30, 20.00, "15,00"),
],
);
xml.push_str("</document>");
xml
}
fn append_page(xml: &mut String, id: &str, lines: &[(f32, f32, &str)]) {
xml.push_str(&format!(r#"<page id="{id}" width="792" height="612">"#));
for (x, y, text) in lines {
append_line(xml, *x, *y, text);
}
xml.push_str("</page>");
}
fn append_line(xml: &mut String, x: f32, y: f32, text: &str) {
let width = x + (text.len() as f32 * 2.0);
let height = y + 3.48;
xml.push_str(&format!(
r#"<line bbox="{x:.2} {y:.2} {width:.2} {height:.2}" text="{}"></line>"#,
escape_xml_attr(text)
));
}
fn escape_xml_attr(text: &str) -> String {
text.chars()
.map(|ch| match ch {
'&' => "&amp;".to_string(),
'<' => "&lt;".to_string(),
'>' => "&gt;".to_string(),
'"' => "&quot;".to_string(),
'\'' => "&apos;".to_string(),
other => other.to_string(),
})
.collect::<String>()
}
}

File diff suppressed because it is too large Load diff

View file

@ -0,0 +1,111 @@
use rusqlite::{Connection, params};
use crate::error::IdxError;
use crate::ownership::types::Entity;
pub fn fts_search(conn: &Connection, query: &str, limit: usize) -> Result<Vec<Entity>, IdxError> {
let q = query.trim();
if q.is_empty() {
return Ok(Vec::new());
}
let limit_i64 =
i64::try_from(limit).map_err(|e| IdxError::DatabaseError(format!("invalid limit: {e}")))?;
// Ensure entity_fts has content. For external-content FTS table, populate manually.
let fts_count: i64 = conn
.query_row("SELECT COUNT(*) FROM entity_fts", [], |row| row.get(0))
.unwrap_or(0);
if fts_count == 0 {
let _ = rebuild_fts(conn);
}
if let Ok(rows) = fts_query(conn, q, limit_i64)
&& !rows.is_empty()
{
return Ok(rows);
}
like_query(conn, q, limit_i64)
}
fn fts_query(conn: &Connection, query: &str, limit: i64) -> Result<Vec<Entity>, IdxError> {
let mut stmt = conn
.prepare(
"SELECT e.id, e.canonical_name, e.entity_type, e.country
FROM entity_fts f
JOIN entities e ON e.id = f.rowid
WHERE entity_fts MATCH ?1
ORDER BY rank
LIMIT ?2",
)
.map_err(|e| IdxError::DatabaseError(e.to_string()))?;
let rows = stmt
.query_map(params![query, limit], |row| {
Ok(Entity {
id: row.get(0)?,
canonical_name: row.get(1)?,
entity_type: row.get(2)?,
country: row.get(3)?,
})
})
.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)
}
fn like_query(conn: &Connection, query: &str, limit: i64) -> Result<Vec<Entity>, IdxError> {
let mut stmt = conn
.prepare(
"SELECT id, canonical_name, entity_type, country
FROM entities
WHERE canonical_name LIKE ?1 COLLATE NOCASE
ORDER BY canonical_name ASC
LIMIT ?2",
)
.map_err(|e| IdxError::DatabaseError(e.to_string()))?;
let like = format!("%{query}%");
let rows = stmt
.query_map(params![like, limit], |row| {
Ok(Entity {
id: row.get(0)?,
canonical_name: row.get(1)?,
entity_type: row.get(2)?,
country: row.get(3)?,
})
})
.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)
}
pub fn rebuild_fts(conn: &Connection) -> Result<(), IdxError> {
conn.execute("DELETE FROM entity_fts", [])
.map_err(|e| IdxError::DatabaseError(e.to_string()))?;
conn.execute(
"INSERT INTO entity_fts(rowid, canonical_name, aliases)
SELECT e.id, e.canonical_name, COALESCE(a.aliases, '')
FROM entities e
LEFT JOIN (
SELECT entity_id, GROUP_CONCAT(raw_name, ' ') AS aliases
FROM entity_aliases
GROUP BY entity_id
) a ON a.entity_id = e.id",
[],
)
.map_err(|e| IdxError::DatabaseError(e.to_string()))?;
Ok(())
}

View file

@ -0,0 +1,894 @@
use std::fs;
use std::path::{Path, PathBuf};
use chrono::NaiveDate;
use rusqlite::{Connection, OpenFlags};
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
use crate::config::get_config_value;
use crate::error::IdxError;
use crate::ownership::db;
use crate::ownership::types::OwnershipRelease;
pub const SNAPSHOT_MANIFEST_CONFIG_KEY: &str = "ownership.snapshot_manifest";
pub const SNAPSHOT_MANIFEST_ENV: &str = "IDX_OWNERSHIP_SNAPSHOT_MANIFEST";
pub const SNAPSHOT_MANIFEST_SCHEMA_VERSION: u32 = 1;
pub const DEFAULT_SNAPSHOT_MANIFEST_URL: &str = "https://github.com/0xrsydn/idx-cli/releases/download/ownership-snapshot-current/ownership-snapshot-manifest.json";
/// ureq caps response bodies at 10 MiB by default. Snapshots with several
/// months of history exceed that (history 3 is ~10.5 MB), so the download is
/// bounded by the manifest's `size_bytes` instead, never above 1 GiB. The body
/// is still verified against that size and the SHA-256 afterwards.
pub(crate) const SNAPSHOT_DOWNLOAD_LIMIT_BYTES: u64 = 1024 * 1024 * 1024;
const USER_AGENT: &str = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36";
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct OwnershipSnapshotManifest {
pub schema_version: u32,
pub generated_at: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub source: Option<OwnershipSnapshotSource>,
pub snapshot: OwnershipSnapshotArtifact,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct OwnershipSnapshotSource {
pub family: String,
pub listing_page_url: String,
pub query_url: String,
pub pdf_url: String,
pub title: String,
pub publish_date: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub original_filename: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct OwnershipSnapshotArtifact {
pub kind: String,
pub compression: String,
pub version: String,
pub download_url: String,
pub sqlite_sha256: String,
pub size_bytes: u64,
pub release_count: usize,
pub latest_as_of_date: NaiveDate,
pub latest_release_sha256: String,
pub latest_row_count: usize,
pub ticker_count: usize,
}
#[derive(Debug, Clone, Copy, Serialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub enum OwnershipSyncAction {
Installed,
Updated,
Refreshed,
NoChange,
SkippedNewer,
SkippedDiverged,
}
#[derive(Debug, Clone, Serialize, PartialEq, Eq)]
pub struct OwnershipSyncResult {
pub action: OwnershipSyncAction,
pub manifest: String,
pub db_path: String,
pub snapshot_version: String,
pub latest_as_of_date: NaiveDate,
pub release_count: usize,
pub ticker_count: usize,
pub reason: String,
}
#[derive(Debug, Clone)]
struct LocalSnapshotState {
latest_release: Option<OwnershipRelease>,
release_count: usize,
}
#[derive(Debug, Clone)]
struct SyncDecision {
action: OwnershipSyncAction,
should_download: bool,
reason: String,
}
pub fn resolve_manifest_source(explicit: Option<&str>) -> Result<String, IdxError> {
let env_value = std::env::var(SNAPSHOT_MANIFEST_ENV).ok();
let config_value = get_config_value(SNAPSHOT_MANIFEST_CONFIG_KEY)?;
resolve_manifest_source_with(explicit, env_value.as_deref(), config_value.as_deref())
}
fn resolve_manifest_source_with(
explicit: Option<&str>,
env_value: Option<&str>,
config_value: Option<&str>,
) -> Result<String, IdxError> {
for value in [explicit, env_value, config_value].into_iter().flatten() {
let trimmed = value.trim();
if !trimmed.is_empty() {
return Ok(trimmed.to_string());
}
}
Ok(DEFAULT_SNAPSHOT_MANIFEST_URL.to_string())
}
pub fn fetch_manifest(source: &str) -> Result<OwnershipSnapshotManifest, IdxError> {
let raw = read_text(source, "ownership snapshot manifest")?;
parse_manifest(&raw)
}
pub fn parse_manifest(raw: &str) -> Result<OwnershipSnapshotManifest, IdxError> {
let manifest: OwnershipSnapshotManifest = serde_json::from_str(raw).map_err(|e| {
IdxError::ParseError(format!(
"failed to parse ownership snapshot manifest JSON: {e}"
))
})?;
validate_manifest(&manifest)?;
Ok(manifest)
}
pub fn sync_snapshot(
manifest_source: &str,
db_path: &Path,
force: bool,
) -> Result<OwnershipSyncResult, IdxError> {
let manifest = fetch_manifest(manifest_source)?;
let local_state = inspect_local_db(db_path)?;
let decision = build_sync_decision(local_state.as_ref(), &manifest.snapshot, force);
if !decision.should_download {
return Ok(OwnershipSyncResult {
action: decision.action,
manifest: manifest_source.to_string(),
db_path: db_path.display().to_string(),
snapshot_version: manifest.snapshot.version.clone(),
latest_as_of_date: manifest.snapshot.latest_as_of_date,
release_count: manifest.snapshot.release_count,
ticker_count: manifest.snapshot.ticker_count,
reason: decision.reason,
});
}
if manifest.snapshot.size_bytes > SNAPSHOT_DOWNLOAD_LIMIT_BYTES {
return Err(IdxError::InvalidInput(format!(
"ownership snapshot manifest reports {} bytes, above the {} byte limit",
manifest.snapshot.size_bytes, SNAPSHOT_DOWNLOAD_LIMIT_BYTES
)));
}
// A body larger than the manifest says is rejected while downloading
// (ureq fails once a body reaches its limit, hence the +1); the exact size
// and SHA-256 are verified afterwards.
let bytes = read_bytes(
&manifest.snapshot.download_url,
"ownership snapshot SQLite",
manifest.snapshot.size_bytes + 1,
)?;
validate_downloaded_bytes(&bytes, &manifest.snapshot)?;
let temp_path = build_temp_path(db_path);
if let Some(parent) = temp_path.parent()
&& !parent.as_os_str().is_empty()
{
fs::create_dir_all(parent).map_err(|e| {
IdxError::Io(format!(
"failed to create snapshot temp directory {}: {e}",
parent.display()
))
})?;
}
fs::write(&temp_path, &bytes).map_err(|e| {
IdxError::Io(format!(
"failed to write downloaded snapshot {}: {e}",
temp_path.display()
))
})?;
let install_result = (|| -> Result<(), IdxError> {
validate_snapshot_db(&temp_path, &manifest.snapshot)?;
install_snapshot_file(&temp_path, db_path)
})();
if install_result.is_err() {
let _ = fs::remove_file(&temp_path);
}
install_result?;
Ok(OwnershipSyncResult {
action: decision.action,
manifest: manifest_source.to_string(),
db_path: db_path.display().to_string(),
snapshot_version: manifest.snapshot.version.clone(),
latest_as_of_date: manifest.snapshot.latest_as_of_date,
release_count: manifest.snapshot.release_count,
ticker_count: manifest.snapshot.ticker_count,
reason: decision.reason,
})
}
fn validate_manifest(manifest: &OwnershipSnapshotManifest) -> Result<(), IdxError> {
if manifest.schema_version != SNAPSHOT_MANIFEST_SCHEMA_VERSION {
return Err(IdxError::ParseError(format!(
"unsupported ownership snapshot manifest schema_version {}; expected {}",
manifest.schema_version, SNAPSHOT_MANIFEST_SCHEMA_VERSION
)));
}
if manifest.generated_at.trim().is_empty() {
return Err(IdxError::ParseError(
"ownership snapshot manifest is missing generated_at".to_string(),
));
}
if let Some(source) = &manifest.source {
if source.family.trim().is_empty() {
return Err(IdxError::ParseError(
"ownership snapshot manifest is missing source.family".to_string(),
));
}
if source.listing_page_url.trim().is_empty() {
return Err(IdxError::ParseError(
"ownership snapshot manifest is missing source.listing_page_url".to_string(),
));
}
if source.query_url.trim().is_empty() {
return Err(IdxError::ParseError(
"ownership snapshot manifest is missing source.query_url".to_string(),
));
}
if source.pdf_url.trim().is_empty() {
return Err(IdxError::ParseError(
"ownership snapshot manifest is missing source.pdf_url".to_string(),
));
}
if source.title.trim().is_empty() {
return Err(IdxError::ParseError(
"ownership snapshot manifest is missing source.title".to_string(),
));
}
if source.publish_date.trim().is_empty() {
return Err(IdxError::ParseError(
"ownership snapshot manifest is missing source.publish_date".to_string(),
));
}
}
let snapshot = &manifest.snapshot;
if snapshot.kind.trim() != "sqlite" {
return Err(IdxError::Unsupported(format!(
"ownership snapshot kind `{}` is not supported; expected `sqlite`",
snapshot.kind
)));
}
if snapshot.compression.trim() != "none" {
return Err(IdxError::Unsupported(format!(
"ownership snapshot compression `{}` is not supported; expected `none`",
snapshot.compression
)));
}
if snapshot.version.trim().is_empty() {
return Err(IdxError::ParseError(
"ownership snapshot manifest is missing snapshot.version".to_string(),
));
}
if snapshot.download_url.trim().is_empty() {
return Err(IdxError::ParseError(
"ownership snapshot manifest is missing snapshot.download_url".to_string(),
));
}
if snapshot.size_bytes == 0 {
return Err(IdxError::ParseError(
"ownership snapshot manifest reported size_bytes=0".to_string(),
));
}
if snapshot.release_count == 0 {
return Err(IdxError::ParseError(
"ownership snapshot manifest reported release_count=0".to_string(),
));
}
if snapshot.latest_row_count == 0 {
return Err(IdxError::ParseError(
"ownership snapshot manifest reported latest_row_count=0".to_string(),
));
}
if snapshot.ticker_count == 0 {
return Err(IdxError::ParseError(
"ownership snapshot manifest reported ticker_count=0".to_string(),
));
}
validate_sha256_hex(&snapshot.sqlite_sha256, "snapshot.sqlite_sha256")?;
validate_sha256_hex(
&snapshot.latest_release_sha256,
"snapshot.latest_release_sha256",
)?;
Ok(())
}
fn validate_sha256_hex(value: &str, field_name: &str) -> Result<(), IdxError> {
let trimmed = value.trim();
if trimmed.len() != 64 || !trimmed.chars().all(|ch| ch.is_ascii_hexdigit()) {
return Err(IdxError::ParseError(format!(
"ownership snapshot manifest field `{field_name}` must be a 64-character hex sha256"
)));
}
Ok(())
}
fn inspect_local_db(db_path: &Path) -> Result<Option<LocalSnapshotState>, IdxError> {
if !db_path.exists() {
return Ok(None);
}
let conn = Connection::open(db_path).map_err(|e| {
IdxError::DatabaseError(format!("failed to open {}: {e}", db_path.display()))
})?;
db::ensure_schema(&conn)?;
let releases = db::query_releases(&conn)?;
Ok(Some(LocalSnapshotState {
latest_release: releases.first().cloned(),
release_count: releases.len(),
}))
}
fn build_sync_decision(
local: Option<&LocalSnapshotState>,
snapshot: &OwnershipSnapshotArtifact,
force: bool,
) -> SyncDecision {
if force {
return SyncDecision {
action: if local.is_some() {
OwnershipSyncAction::Refreshed
} else {
OwnershipSyncAction::Installed
},
should_download: true,
reason: "force refresh requested".to_string(),
};
}
let Some(local) = local else {
return SyncDecision {
action: OwnershipSyncAction::Installed,
should_download: true,
reason: "local ownership database does not exist yet".to_string(),
};
};
let Some(latest) = &local.latest_release else {
return SyncDecision {
action: OwnershipSyncAction::Updated,
should_download: true,
reason: "local ownership database exists but has no imported releases".to_string(),
};
};
if latest.as_of_date > snapshot.latest_as_of_date {
return SyncDecision {
action: OwnershipSyncAction::SkippedNewer,
should_download: false,
reason: format!(
"local ownership database is newer than the published snapshot (local {} > snapshot {})",
latest.as_of_date.format("%Y-%m-%d"),
snapshot.latest_as_of_date.format("%Y-%m-%d")
),
};
}
if latest.as_of_date < snapshot.latest_as_of_date {
return SyncDecision {
action: OwnershipSyncAction::Updated,
should_download: true,
reason: format!(
"published snapshot advances local ownership data from {} to {}",
latest.as_of_date.format("%Y-%m-%d"),
snapshot.latest_as_of_date.format("%Y-%m-%d")
),
};
}
if latest.sha256 == snapshot.latest_release_sha256 {
if latest.row_count != snapshot.latest_row_count {
return SyncDecision {
action: OwnershipSyncAction::Updated,
should_download: true,
reason: "local ownership database has the same latest release date but different row_count metadata".to_string(),
};
}
if local.release_count < snapshot.release_count {
return SyncDecision {
action: OwnershipSyncAction::Updated,
should_download: true,
reason: format!(
"local ownership database is missing historical releases (local {} < snapshot {})",
local.release_count, snapshot.release_count
),
};
}
if local.release_count == snapshot.release_count {
return SyncDecision {
action: OwnershipSyncAction::NoChange,
should_download: false,
reason: format!(
"ownership snapshot already current at {} with {} release(s)",
snapshot.latest_as_of_date.format("%Y-%m-%d"),
snapshot.release_count
),
};
}
}
SyncDecision {
action: OwnershipSyncAction::SkippedDiverged,
should_download: false,
reason: format!(
"local ownership database differs from the published snapshot for {}; use --force to replace it",
snapshot.latest_as_of_date.format("%Y-%m-%d")
),
}
}
fn validate_downloaded_bytes(
bytes: &[u8],
snapshot: &OwnershipSnapshotArtifact,
) -> Result<(), IdxError> {
let size_bytes = u64::try_from(bytes.len()).map_err(|e| {
IdxError::ParseError(format!("snapshot download too large to validate: {e}"))
})?;
if size_bytes != snapshot.size_bytes {
return Err(IdxError::ParseError(format!(
"ownership snapshot size mismatch: manifest expected {} bytes, downloaded {} bytes",
snapshot.size_bytes, size_bytes
)));
}
let actual_sha256 = sha256_hex(bytes);
if actual_sha256 != snapshot.sqlite_sha256 {
return Err(IdxError::ParseError(format!(
"ownership snapshot checksum mismatch: manifest expected {}, downloaded {}",
snapshot.sqlite_sha256, actual_sha256
)));
}
Ok(())
}
fn validate_snapshot_db(path: &Path, snapshot: &OwnershipSnapshotArtifact) -> Result<(), IdxError> {
let conn =
Connection::open_with_flags(path, OpenFlags::SQLITE_OPEN_READ_ONLY).map_err(|e| {
IdxError::DatabaseError(format!(
"failed to open downloaded snapshot {}: {e}",
path.display()
))
})?;
let quick_check: String = conn
.query_row("PRAGMA quick_check(1)", [], |row| row.get(0))
.map_err(|e| IdxError::DatabaseError(format!("snapshot quick_check failed: {e}")))?;
if quick_check.trim() != "ok" {
return Err(IdxError::DatabaseError(format!(
"snapshot quick_check failed: {quick_check}"
)));
}
let releases = db::query_releases(&conn)?;
if releases.len() != snapshot.release_count {
return Err(IdxError::ParseError(format!(
"downloaded snapshot release_count mismatch: manifest expected {}, sqlite has {}",
snapshot.release_count,
releases.len()
)));
}
let latest = releases.first().ok_or_else(|| {
IdxError::ParseError("downloaded snapshot sqlite had no ownership releases".to_string())
})?;
if latest.as_of_date != snapshot.latest_as_of_date {
return Err(IdxError::ParseError(format!(
"downloaded snapshot latest_as_of_date mismatch: manifest expected {}, sqlite has {}",
snapshot.latest_as_of_date.format("%Y-%m-%d"),
latest.as_of_date.format("%Y-%m-%d")
)));
}
if latest.sha256 != snapshot.latest_release_sha256 {
return Err(IdxError::ParseError(format!(
"downloaded snapshot latest_release_sha256 mismatch: manifest expected {}, sqlite has {}",
snapshot.latest_release_sha256, latest.sha256
)));
}
if latest.row_count != snapshot.latest_row_count {
return Err(IdxError::ParseError(format!(
"downloaded snapshot latest_row_count mismatch: manifest expected {}, sqlite has {}",
snapshot.latest_row_count, latest.row_count
)));
}
let ticker_count: i64 = conn
.query_row("SELECT COUNT(*) FROM tickers", [], |row| row.get(0))
.map_err(|e| IdxError::DatabaseError(format!("failed counting snapshot tickers: {e}")))?;
let ticker_count = usize::try_from(ticker_count)
.map_err(|e| IdxError::DatabaseError(format!("invalid snapshot ticker_count: {e}")))?;
if ticker_count != snapshot.ticker_count {
return Err(IdxError::ParseError(format!(
"downloaded snapshot ticker_count mismatch: manifest expected {}, sqlite has {}",
snapshot.ticker_count, ticker_count
)));
}
Ok(())
}
fn build_temp_path(db_path: &Path) -> PathBuf {
let file_name = db_path
.file_name()
.and_then(|name| name.to_str())
.unwrap_or("ownership.db");
let temp_name = format!(".{file_name}.sync-{}.tmp", fastrand::u64(..));
match db_path.parent() {
Some(parent) if !parent.as_os_str().is_empty() => parent.join(temp_name),
_ => PathBuf::from(temp_name),
}
}
fn install_snapshot_file(temp_path: &Path, db_path: &Path) -> Result<(), IdxError> {
if let Some(parent) = db_path.parent()
&& !parent.as_os_str().is_empty()
{
fs::create_dir_all(parent).map_err(|e| {
IdxError::Io(format!(
"failed to create ownership database directory {}: {e}",
parent.display()
))
})?;
}
let backup_path = build_backup_path(db_path);
if backup_path.exists() {
fs::remove_file(&backup_path).map_err(|e| {
IdxError::Io(format!(
"failed to remove stale backup {}: {e}",
backup_path.display()
))
})?;
}
let had_existing = db_path.exists();
remove_sqlite_sidecars(db_path)?;
if had_existing {
fs::rename(db_path, &backup_path).map_err(|e| {
IdxError::Io(format!(
"failed to move existing ownership database {} to backup {}: {e}",
db_path.display(),
backup_path.display()
))
})?;
}
if let Err(err) = fs::rename(temp_path, db_path) {
if had_existing {
let _ = fs::rename(&backup_path, db_path);
}
return Err(IdxError::Io(format!(
"failed to install ownership snapshot into {}: {err}",
db_path.display()
)));
}
if had_existing && backup_path.exists() {
fs::remove_file(&backup_path).map_err(|e| {
IdxError::Io(format!(
"failed to remove snapshot backup {}: {e}",
backup_path.display()
))
})?;
}
remove_sqlite_sidecars(db_path)?;
Ok(())
}
fn build_backup_path(db_path: &Path) -> PathBuf {
let file_name = db_path
.file_name()
.and_then(|name| name.to_str())
.unwrap_or("ownership.db");
let backup_name = format!(".{file_name}.sync-backup");
match db_path.parent() {
Some(parent) if !parent.as_os_str().is_empty() => parent.join(backup_name),
_ => PathBuf::from(backup_name),
}
}
fn remove_sqlite_sidecars(db_path: &Path) -> Result<(), IdxError> {
for suffix in ["-wal", "-shm"] {
let file_name = format!(
"{}{suffix}",
db_path
.file_name()
.and_then(|name| name.to_str())
.unwrap_or("ownership.db")
);
let sidecar = match db_path.parent() {
Some(parent) if !parent.as_os_str().is_empty() => parent.join(&file_name),
_ => PathBuf::from(&file_name),
};
if sidecar.exists() {
fs::remove_file(&sidecar).map_err(|e| {
IdxError::Io(format!(
"failed to remove SQLite sidecar {}: {e}",
sidecar.display()
))
})?;
}
}
Ok(())
}
fn sha256_hex(bytes: &[u8]) -> String {
let mut hasher = Sha256::new();
hasher.update(bytes);
let digest = hasher.finalize();
digest.iter().map(|byte| format!("{byte:02x}")).collect()
}
fn read_text(source: &str, context: &str) -> Result<String, IdxError> {
if is_http_source(source) {
let response = ureq::get(source)
.header("User-Agent", USER_AGENT)
.header("Accept", "application/json,text/plain;q=0.9,*/*;q=0.8")
.call()
.map_err(|e| IdxError::Http(format!("failed to fetch {context}: {e}")))?;
let mut body = response.into_body();
return body
.read_to_string()
.map_err(|e| IdxError::Http(format!("failed reading {context} body: {e}")));
}
let path = local_source_path(source);
fs::read_to_string(&path)
.map_err(|e| IdxError::Io(format!("failed to read {context} {}: {e}", path.display())))
}
fn read_bytes(source: &str, context: &str, limit_bytes: u64) -> Result<Vec<u8>, IdxError> {
if is_http_source(source) {
let response = ureq::get(source)
.header("User-Agent", USER_AGENT)
.header(
"Accept",
"application/octet-stream,application/x-sqlite3,*/*;q=0.8",
)
.call()
.map_err(|e| IdxError::Http(format!("failed to fetch {context}: {e}")))?;
return response
.into_body()
.with_config()
.limit(limit_bytes)
.read_to_vec()
.map_err(|e| {
IdxError::Http(format!(
"failed reading {context} body (larger than the manifest's size_bytes?): {e}"
))
});
}
let path = local_source_path(source);
fs::read(&path)
.map_err(|e| IdxError::Io(format!("failed to read {context} {}: {e}", path.display())))
}
fn is_http_source(source: &str) -> bool {
let normalized = source.trim().to_ascii_lowercase();
normalized.starts_with("http://") || normalized.starts_with("https://")
}
fn local_source_path(source: &str) -> PathBuf {
if let Some(path) = source.trim().strip_prefix("file://") {
return PathBuf::from(path);
}
PathBuf::from(source.trim())
}
#[cfg(test)]
mod tests {
use chrono::NaiveDate;
use rusqlite::Connection;
use super::{
DEFAULT_SNAPSHOT_MANIFEST_URL, OwnershipSnapshotArtifact, OwnershipSnapshotManifest,
OwnershipSnapshotSource, OwnershipSyncAction, SNAPSHOT_MANIFEST_SCHEMA_VERSION,
build_sync_decision, parse_manifest, resolve_manifest_source_with,
};
use crate::ownership::db::{ensure_schema, insert_release};
use crate::ownership::types::OwnershipRelease;
#[test]
fn parse_manifest_rejects_unknown_schema_version() {
let raw = r#"{
"schema_version": 99,
"generated_at": "2026-03-31T12:00:00Z",
"snapshot": {
"kind": "sqlite",
"compression": "none",
"version": "2026-02-27",
"download_url": "/tmp/ownership.sqlite",
"sqlite_sha256": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
"size_bytes": 123,
"release_count": 2,
"latest_as_of_date": "2026-02-27",
"latest_release_sha256": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb",
"latest_row_count": 100,
"ticker_count": 5
}
}"#;
let err = parse_manifest(raw).expect_err("schema_version should fail");
assert!(
err.to_string()
.contains("unsupported ownership snapshot manifest schema_version")
);
}
#[test]
fn build_sync_decision_updates_same_latest_release_when_history_is_incomplete() {
let snapshot = OwnershipSnapshotArtifact {
kind: "sqlite".to_string(),
compression: "none".to_string(),
version: "2026-02-27".to_string(),
download_url: "/tmp/ownership.sqlite".to_string(),
sqlite_sha256: "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
.to_string(),
size_bytes: 123,
release_count: 2,
latest_as_of_date: NaiveDate::from_ymd_opt(2026, 2, 27).unwrap(),
latest_release_sha256:
"bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb".to_string(),
latest_row_count: 7261,
ticker_count: 955,
};
let local = super::LocalSnapshotState {
latest_release: Some(OwnershipRelease {
id: 1,
source_url: None,
sha256: snapshot.latest_release_sha256.clone(),
as_of_date: snapshot.latest_as_of_date,
row_count: snapshot.latest_row_count,
imported_at: 0,
}),
release_count: 1,
};
let decision = build_sync_decision(Some(&local), &snapshot, false);
assert_eq!(decision.action, OwnershipSyncAction::Updated);
assert!(decision.should_download);
assert!(decision.reason.contains("missing historical releases"));
}
#[test]
fn manifest_round_trip_parses_valid_payload() {
let manifest = OwnershipSnapshotManifest {
schema_version: SNAPSHOT_MANIFEST_SCHEMA_VERSION,
generated_at: "2026-03-31T12:00:00Z".to_string(),
source: Some(OwnershipSnapshotSource {
family: "above1".to_string(),
listing_page_url: "https://www.idx.co.id/id/berita/pengumuman/".to_string(),
query_url: "https://www.idx.co.id/primary/NewsAnnouncement/GetAllAnnouncement?keywords=pemegang%20saham%20di%20atas%201&pageNumber=1&pageSize=10&lang=id".to_string(),
pdf_url: "https://www.idx.co.id/StaticData/sample.pdf".to_string(),
title: "Pemegang Saham di atas 1% (KSEI)".to_string(),
publish_date: "2026-03-10T00:00:00".to_string(),
original_filename: Some("sample.pdf".to_string()),
}),
snapshot: OwnershipSnapshotArtifact {
kind: "sqlite".to_string(),
compression: "none".to_string(),
version: "2026-02-27".to_string(),
download_url: "/tmp/ownership.sqlite".to_string(),
sqlite_sha256: "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
.to_string(),
size_bytes: 123,
release_count: 1,
latest_as_of_date: NaiveDate::from_ymd_opt(2026, 2, 27).unwrap(),
latest_release_sha256:
"bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb".to_string(),
latest_row_count: 7261,
ticker_count: 955,
},
};
let raw = serde_json::to_string(&manifest).unwrap();
let parsed = parse_manifest(&raw).expect("valid manifest");
assert_eq!(parsed, manifest);
}
#[test]
fn manifest_without_source_metadata_still_parses() {
let raw = r#"{
"schema_version": 1,
"generated_at": "2026-03-31T12:00:00Z",
"snapshot": {
"kind": "sqlite",
"compression": "none",
"version": "2026-02-27",
"download_url": "/tmp/ownership.sqlite",
"sqlite_sha256": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
"size_bytes": 123,
"release_count": 1,
"latest_as_of_date": "2026-02-27",
"latest_release_sha256": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb",
"latest_row_count": 7261,
"ticker_count": 955
}
}"#;
let parsed = parse_manifest(raw).expect("old manifest shape should still parse");
assert!(parsed.source.is_none());
}
#[test]
fn resolve_manifest_source_uses_built_in_default_when_unset() {
let resolved = resolve_manifest_source_with(None, None, None).expect("default manifest");
assert_eq!(resolved, DEFAULT_SNAPSHOT_MANIFEST_URL);
}
#[test]
fn resolve_manifest_source_prefers_explicit_then_env_then_config() {
let explicit = resolve_manifest_source_with(
Some("https://example.com/explicit.json"),
Some("https://example.com/env.json"),
Some("https://example.com/config.json"),
)
.expect("explicit manifest");
assert_eq!(explicit, "https://example.com/explicit.json");
let env = resolve_manifest_source_with(
Some(" "),
Some("https://example.com/env.json"),
Some("https://example.com/config.json"),
)
.expect("env manifest");
assert_eq!(env, "https://example.com/env.json");
let config =
resolve_manifest_source_with(None, Some(" "), Some("https://example.com/config.json"))
.expect("config manifest");
assert_eq!(config, "https://example.com/config.json");
}
#[test]
fn ensure_schema_can_store_release_metadata_needed_for_snapshots() {
let conn = Connection::open_in_memory().unwrap();
ensure_schema(&conn).unwrap();
insert_release(
&conn,
&OwnershipRelease {
id: 0,
source_url: Some("https://example.com/ownership.sqlite".to_string()),
sha256: "abc".to_string(),
as_of_date: NaiveDate::from_ymd_opt(2026, 2, 27).unwrap(),
row_count: 1,
imported_at: 1,
},
)
.unwrap();
let release_count: i64 = conn
.query_row("SELECT COUNT(*) FROM ownership_releases", [], |row| {
row.get(0)
})
.unwrap();
assert_eq!(release_count, 1);
}
}

View file

@ -0,0 +1,451 @@
use chrono::NaiveDate;
use serde::{Deserialize, Serialize};
/// Which data source a holding originates from.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum OwnershipSource {
/// KSEI shareholder registry source.
Ksei,
/// Bing institutional ownership source.
Bing,
}
/// Investor locality classification from KSEI.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub enum Locality {
/// Local Indonesian investor.
Local,
/// Foreign investor.
Foreign,
}
/// Bing institutional flow signal.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum FlowSignal {
/// Existing top holder.
Holder,
/// Net buyer for the period.
Buyer,
/// Net seller for the period.
Seller,
/// New position opened this period.
NewPosition,
/// Fully exited this period.
Exited,
}
/// Method used to resolve an entity alias to a canonical entity.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum ResolutionMethod {
/// Exact string match after normalization.
Exact,
/// Rule-based normalization match.
Rule,
/// Fuzzy or similarity-based match.
Fuzzy,
/// Human-curated manual mapping.
Manual,
}
/// Graph node category used in ownership network output.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum GraphNodeType {
/// Canonical investor entity node.
Entity,
/// Listed issuer ticker node.
Ticker,
}
/// KSEI investor type code (for example: `CP`, `ID`, `IB`, `MF`, `SC`, `IS`, `OT`).
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct InvestorTypeCode(pub String);
/// Canonical resolved entity (investor or shareholder).
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Entity {
/// Internal entity identifier.
pub id: i64,
/// Canonical normalized display name.
pub canonical_name: String,
/// Optional coarse entity type (`fund`, `bank`, `conglomerate`, `govt`, `individual`).
pub entity_type: Option<String>,
/// Optional ISO-like country tag.
pub country: Option<String>,
}
/// A raw name variant mapped to a canonical entity.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct EntityAlias {
/// Internal alias identifier.
pub id: i64,
/// Referenced canonical entity id.
pub entity_id: i64,
/// Raw alias text from source.
pub raw_name: String,
/// Source that produced this alias.
pub source: OwnershipSource,
/// Resolution confidence score in `0.0..=1.0`.
pub confidence: f64,
/// Matching method used for resolution.
pub method: ResolutionMethod,
}
/// Ticker (issuer) reference metadata.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Ticker {
/// Internal ticker identifier.
pub id: i64,
/// Exchange ticker code, for example `BBCA`.
pub code: String,
/// Optional long issuer name.
pub name: Option<String>,
/// Optional sector classification.
pub sector: Option<String>,
}
/// Single KSEI ownership row for one holder in one ticker and release.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct KseiHolding {
/// Internal holding row identifier.
pub id: i64,
/// Referenced ticker id.
pub ticker_id: i64,
/// Referenced entity id, unresolved when `None`.
pub entity_id: Option<i64>,
/// Raw investor name exactly as imported.
pub raw_investor_name: String,
/// Optional KSEI investor type code.
pub investor_type: Option<InvestorTypeCode>,
/// Optional local/foreign classification.
pub locality: Option<Locality>,
/// Optional nationality text.
pub nationality: Option<String>,
/// Optional domicile text.
pub domicile: Option<String>,
/// Scripless holdings in shares.
pub holdings_scripless: i64,
/// Script holdings in shares.
pub holdings_scrip: i64,
/// Total shares held.
pub total_shares: i64,
/// Ownership percentage in basis points (`41.10% -> 4110`).
pub percentage_bps: i64,
/// Snapshot as-of date.
pub report_date: NaiveDate,
/// SHA-256 hash of source release for deduplication.
pub release_sha256: String,
}
/// Draft holding before entity resolution (without persisted row id and entity id).
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct KseiHoldingDraft {
/// Exchange ticker code, for example `BBCA`.
pub ticker_code: String,
/// Optional issuer name as provided by source row.
pub issuer_name: Option<String>,
/// Raw investor name exactly as imported.
pub raw_investor_name: String,
/// Optional KSEI investor type code.
pub investor_type: Option<InvestorTypeCode>,
/// Optional local/foreign classification.
pub locality: Option<Locality>,
/// Optional nationality text.
pub nationality: Option<String>,
/// Optional domicile text.
pub domicile: Option<String>,
/// Scripless holdings in shares.
pub holdings_scripless: i64,
/// Script holdings in shares.
pub holdings_scrip: i64,
/// Total shares held.
pub total_shares: i64,
/// Ownership percentage in basis points (`41.10% -> 4110`).
pub percentage_bps: i64,
/// Snapshot as-of date.
pub report_date: NaiveDate,
}
/// Single Bing institutional ownership row.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct BingHolding {
/// Internal holding row identifier.
pub id: i64,
/// Referenced ticker id.
pub ticker_id: i64,
/// Referenced entity id, unresolved when `None`.
pub entity_id: Option<i64>,
/// Raw investor name exactly as imported.
pub raw_investor_name: String,
/// Optional Bing investor type text.
pub investor_type: Option<String>,
/// Shares currently held.
pub shares_held: Option<i64>,
/// Share delta over period (`+buy`, `-sell`).
pub shares_changed: Option<i64>,
/// Ownership percentage in basis points.
pub pct_ownership_bps: Option<i64>,
/// Position value in USD.
pub value_usd: Option<i64>,
/// Source report date.
pub report_date: NaiveDate,
/// Institutional flow signal for the row.
pub signal: FlowSignal,
/// Import timestamp (unix epoch seconds).
pub fetched_at: i64,
}
/// Raw row extracted from KSEI PDF before normalization.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct KseiRawRow {
/// Date string as appears in PDF (for example `27-Feb-2026`).
pub date: String,
/// Share code string (for example `BBCA`).
pub share_code: String,
/// Issuer name as appears in source.
pub issuer_name: String,
/// Investor name as appears in source.
pub investor_name: String,
/// Investor type code string.
pub investor_type: String,
/// Local or foreign marker (`L` or `F`).
pub local_foreign: String,
/// Nationality text.
pub nationality: String,
/// Domicile text.
pub domicile: String,
/// Scripless holdings string in Indonesian locale format.
pub holdings_scripless: String,
/// Script holdings string in Indonesian locale format.
pub holdings_scrip: String,
/// Total holdings string in Indonesian locale format.
pub total_holding_shares: String,
/// Percentage string in Indonesian locale decimal format.
pub percentage: String,
}
/// Raw holder object returned by Bing ownership endpoints.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct BingHolderRaw {
/// Investor name field from Bing payload.
#[serde(alias = "investorName", alias = "InvestorName")]
pub investor_name: Option<String>,
/// Investor type field from Bing payload.
#[serde(alias = "investorType", alias = "InvestorType")]
pub investor_type: Option<String>,
/// Shares held field from Bing payload.
#[serde(alias = "sharesHeld", alias = "SharesHeld")]
pub shares_held: Option<f64>,
/// Shares changed field from Bing payload.
#[serde(alias = "sharesChanged", alias = "SharesChanged")]
pub shares_changed: Option<f64>,
/// Percentage of shares outstanding field from Bing payload.
#[serde(
alias = "percentageOfSharesOutstanding",
alias = "PercentageOfSharesOutstanding"
)]
pub pct_outstanding: Option<f64>,
/// Position value field from Bing payload.
#[serde(alias = "value", alias = "Value")]
pub value: Option<f64>,
/// Report date field from Bing payload.
#[serde(alias = "reportDate", alias = "ReportDate")]
pub report_date: Option<String>,
}
/// Combined ownership view for a ticker (KSEI and Bing merged).
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TickerOwnership {
/// Ticker metadata.
pub ticker: Ticker,
/// Latest KSEI as-of date.
pub ksei_as_of: Option<NaiveDate>,
/// Latest Bing reporting period label.
pub bing_as_of: Option<String>,
/// Combined holder rows for display.
pub holders: Vec<HolderRow>,
/// Concentration metrics calculated for the ticker.
pub concentration: ConcentrationMetrics,
/// Optional institutional flow breakdown.
pub flow: Option<InstitutionalFlow>,
}
/// Single row in a combined holders table.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct HolderRow {
/// Ranking position in result set.
pub rank: usize,
/// Source dataset for the row.
pub source: OwnershipSource,
/// Canonical holder name or unresolved raw name.
pub name: String,
/// Optional referenced canonical entity id.
pub entity_id: Option<i64>,
/// Optional investor type label.
pub investor_type: Option<String>,
/// Optional local/foreign marker.
pub locality: Option<Locality>,
/// Held shares quantity.
pub shares: i64,
/// Ownership percentage in basis points.
pub percentage_bps: i64,
/// Optional flow signal (Bing rows only).
pub signal: Option<FlowSignal>,
}
/// Concentration metrics for ownership distribution of a ticker.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ConcentrationMetrics {
/// Largest single holder percentage in basis points.
pub top1_bps: i64,
/// Sum of top 3 holder percentages in basis points.
pub top3_bps: i64,
/// Herfindahl-Hirschman index value.
pub hhi: i64,
/// Estimated free float in basis points (`10000 - known holders`).
pub free_float_bps: i64,
/// Number of KSEI holders above or equal to 1%.
pub holder_count: usize,
}
/// Institutional flow summary from Bing for one ticker and period.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct InstitutionalFlow {
/// Reporting period label.
pub period: String,
/// Top buyer rows.
pub top_buyers: Vec<HolderRow>,
/// Top seller rows.
pub top_sellers: Vec<HolderRow>,
/// New position rows.
pub new_positions: Vec<HolderRow>,
/// Exited position rows.
pub exited: Vec<HolderRow>,
}
/// Cross-holding summary for a single entity.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct EntityHoldings {
/// Canonical entity metadata.
pub entity: Entity,
/// Number of distinct tickers held.
pub ticker_count: usize,
/// Per-ticker ownership rows.
pub holdings: Vec<EntityTickerRow>,
}
/// One ticker row within an entity holdings summary.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct EntityTickerRow {
/// Ticker metadata.
pub ticker: Ticker,
/// Source dataset for the holding.
pub source: OwnershipSource,
/// Shares held quantity.
pub shares: i64,
/// Ownership percentage in basis points.
pub percentage_bps: i64,
/// Report as-of date.
pub report_date: NaiveDate,
}
/// Cross-holder leaderboard row across multiple tickers.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CrossHolderRow {
/// Canonical entity metadata.
pub entity: Entity,
/// Number of distinct tickers held.
pub ticker_count: usize,
/// Summed ownership basis points across tickers.
pub total_bps: i64,
}
/// Graph node used for ownership network visualization.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct GraphNode {
/// Stable graph node id (`entity:42` or `ticker:BBCA`).
pub id: String,
/// Display label.
pub label: String,
/// Node category.
pub node_type: GraphNodeType,
}
/// Graph edge used for ownership network visualization.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct GraphEdge {
/// Source node id.
pub from: String,
/// Target node id.
pub to: String,
/// Edge weight in basis points.
pub percentage_bps: i64,
/// Source dataset of the edge.
pub source: OwnershipSource,
}
/// Imported ownership release metadata.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct OwnershipRelease {
/// Internal release id.
pub id: i64,
/// Optional source URL where release was obtained.
pub source_url: Option<String>,
/// SHA-256 hash for release-level deduplication.
pub sha256: String,
/// Release as-of date.
pub as_of_date: NaiveDate,
/// Parsed row count imported.
pub row_count: usize,
/// Import timestamp (unix epoch seconds).
pub imported_at: i64,
}
/// Type of ownership change between two snapshots.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum ChangeType {
/// Holder appears in `to` snapshot but not in `from`.
New,
/// Holder appears in `from` snapshot but not in `to`.
Exited,
/// Holder exists in both snapshots and percentage increased.
Increased,
/// Holder exists in both snapshots and percentage decreased.
Decreased,
}
/// One ownership change row for a ticker-holder pair.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ChangeRow {
/// Exchange ticker code.
pub ticker_code: String,
/// Canonical or raw holder name.
pub entity_name: String,
/// Change classification.
pub change_type: ChangeType,
/// Old percentage in basis points (from snapshot).
pub old_bps: Option<i64>,
/// New percentage in basis points (to snapshot).
pub new_bps: Option<i64>,
/// Delta in basis points (`new - old`, missing treated as 0).
pub delta_bps: i64,
}
/// Row describing unresolved or low-confidence alias mappings.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct UnresolvedRow {
/// Raw investor name captured from source data.
pub raw_name: String,
/// Source marker (`ksei` or `bing`).
pub source: String,
/// Ticker code where the unresolved alias appears.
pub ticker_code: String,
/// Current canonical entity name when available.
pub current_entity: Option<String>,
/// Resolution confidence score.
pub confidence: Option<f64>,
}

View file

@ -0,0 +1,554 @@
//! Parser for the IDX "Pemegang Saham di Atas 1%" XLSX workbook.
//!
//! Since June 2026 IDX publishes the monthly above-1% holder register as an
//! XLSX file on the "Data Kepemilikan Saham" page instead of a PDF
//! announcement. The workbook has one sheet: a disclaimer block, a `[PUBLIC]`
//! marker, then a fixed 12-column table. Only that exact header is accepted,
//! so the daily above-5% workbook and the investor-type / classification
//! workbooks published on the same page are rejected instead of mis-imported.
use std::collections::HashMap;
use std::fs;
use std::io::{Cursor, Read};
use std::path::Path;
use chrono::{Duration, NaiveDate};
use quick_xml::Reader;
use quick_xml::events::{BytesStart, Event};
use crate::error::IdxError;
use crate::ownership::entities::{normalize_investor_type, normalize_locality, parse_ksei_date};
use crate::ownership::types::KseiHoldingDraft;
const EXPECTED_HEADER: [&str; 12] = [
"DATE",
"SHARE_CODE",
"ISSUER_NAME",
"INVESTOR_NAME",
"INVESTOR_CLASSIFICATION",
"LOCAL_FOREIGN",
"NATIONALITY",
"DOMICILE",
"HOLDINGS_SCRIPLESS",
"HOLDINGS_SCRIP",
"TOTAL_HOLDING_SHARES",
"PERCENTAGE",
];
/// Rows above this line are disclaimer text; the header must appear before it.
const MAX_HEADER_SEARCH_ROWS: usize = 50;
pub fn supports_xlsx_file(path: &Path) -> bool {
path.extension()
.and_then(|value| value.to_str())
.is_some_and(|value| value.trim().eq_ignore_ascii_case("xlsx"))
}
pub fn parse_above1_xlsx_file(path: &Path) -> Result<Vec<KseiHoldingDraft>, IdxError> {
let bytes = fs::read(path).map_err(|e| {
IdxError::Io(format!(
"failed to read ownership XLSX {}: {e}",
path.display()
))
})?;
parse_above1_xlsx_bytes(&bytes)
}
pub fn parse_above1_xlsx_bytes(bytes: &[u8]) -> Result<Vec<KseiHoldingDraft>, IdxError> {
let mut zip = zip::ZipArchive::new(Cursor::new(bytes))
.map_err(|e| IdxError::ParseError(format!("failed to open ownership XLSX: {e}")))?;
let shared_strings = match read_zip_entry(&mut zip, "xl/sharedStrings.xml")? {
Some(xml) => parse_shared_strings(&xml)?,
None => Vec::new(),
};
let sheet_name = first_worksheet_name(&mut zip)?;
let sheet_xml = read_zip_entry(&mut zip, &sheet_name)?
.ok_or_else(|| IdxError::ParseError(format!("ownership XLSX is missing {sheet_name}")))?;
let rows = parse_sheet_rows(&sheet_xml, &shared_strings)?;
drafts_from_rows(&rows)
}
fn read_zip_entry(
zip: &mut zip::ZipArchive<Cursor<&[u8]>>,
name: &str,
) -> Result<Option<String>, IdxError> {
let mut file = match zip.by_name(name) {
Ok(file) => file,
Err(zip::result::ZipError::FileNotFound) => return Ok(None),
Err(e) => {
return Err(IdxError::ParseError(format!(
"failed to read ownership XLSX entry {name}: {e}"
)));
}
};
let mut output = String::new();
file.read_to_string(&mut output).map_err(|e| {
IdxError::ParseError(format!("failed to decode ownership XLSX entry {name}: {e}"))
})?;
Ok(Some(output))
}
fn first_worksheet_name(zip: &mut zip::ZipArchive<Cursor<&[u8]>>) -> Result<String, IdxError> {
let mut sheets: Vec<String> = zip
.file_names()
.filter(|name| name.starts_with("xl/worksheets/") && name.ends_with(".xml"))
.filter(|name| !name["xl/worksheets/".len()..].contains('/'))
.map(str::to_string)
.collect();
// sheet1.xml, sheet2.xml, ... sort by their numeric suffix.
sheets.sort_by_key(|name| {
name.trim_start_matches("xl/worksheets/sheet")
.trim_end_matches(".xml")
.parse::<u32>()
.unwrap_or(u32::MAX)
});
sheets
.into_iter()
.next()
.ok_or_else(|| IdxError::ParseError("ownership XLSX has no worksheet".to_string()))
}
fn xml_error(context: &str, err: impl std::fmt::Display) -> IdxError {
IdxError::ParseError(format!("failed to parse ownership XLSX {context}: {err}"))
}
/// Shared strings table: each `<si>` is one string, made of one `<t>` or
/// several rich-text runs `<r><t>`. Phonetic hints (`<rPh>`) are skipped.
fn parse_shared_strings(xml: &str) -> Result<Vec<String>, IdxError> {
let mut reader = Reader::from_str(xml);
let mut strings = Vec::new();
let mut current: Option<String> = None;
let mut in_text = false;
let mut in_phonetic = false;
loop {
match reader.read_event() {
Ok(Event::Start(e)) => match e.local_name().as_ref() {
b"si" => current = Some(String::new()),
b"t" => in_text = true,
b"rPh" => in_phonetic = true,
_ => {}
},
Ok(Event::Empty(e)) if e.local_name().as_ref() == b"si" => strings.push(String::new()),
Ok(Event::Text(text)) if in_text && !in_phonetic => {
if let Some(value) = current.as_mut() {
let unescaped = text
.unescape()
.map_err(|e| xml_error("shared strings", e))?;
value.push_str(&unescaped);
}
}
Ok(Event::End(e)) => match e.local_name().as_ref() {
b"si" => strings.push(current.take().unwrap_or_default()),
b"t" => in_text = false,
b"rPh" => in_phonetic = false,
_ => {}
},
Ok(Event::Eof) => break,
Err(e) => return Err(xml_error("shared strings", e)),
_ => {}
}
}
Ok(strings)
}
/// One sheet row: its 1-based Excel row number and column letters
/// (`A`, `B`, ...) mapped to cell text. Sheet XML omits blank rows, so the
/// number comes from the row's `r` attribute, not its position.
#[derive(Debug, Default)]
struct SheetRow {
number: usize,
cells: HashMap<String, String>,
}
fn row_number(element: &BytesStart<'_>, fallback: usize) -> Result<usize, IdxError> {
for attribute in element.attributes() {
let attribute = attribute.map_err(|e| xml_error("worksheet", e))?;
if attribute.key.local_name().as_ref() == b"r" {
if let Ok(number) = String::from_utf8_lossy(&attribute.value).trim().parse() {
return Ok(number);
}
}
}
Ok(fallback)
}
fn parse_sheet_rows(xml: &str, shared_strings: &[String]) -> Result<Vec<SheetRow>, IdxError> {
let mut reader = Reader::from_str(xml);
let mut rows = Vec::new();
let mut row: Option<SheetRow> = None;
let mut cell: Option<(String, String)> = None;
let mut value = String::new();
let mut in_value = false;
loop {
match reader.read_event() {
Ok(Event::Start(e)) => match e.local_name().as_ref() {
b"row" => {
row = Some(SheetRow {
number: row_number(&e, rows.len() + 1)?,
cells: HashMap::new(),
})
}
b"c" => {
cell = Some(cell_ref_and_type(&e)?);
value.clear();
}
b"v" | b"t" if cell.is_some() => in_value = true,
_ => {}
},
Ok(Event::Empty(e)) if e.local_name().as_ref() == b"row" => rows.push(SheetRow {
number: row_number(&e, rows.len() + 1)?,
cells: HashMap::new(),
}),
Ok(Event::Text(text)) if in_value => {
let unescaped = text.unescape().map_err(|e| xml_error("worksheet", e))?;
value.push_str(&unescaped);
}
Ok(Event::End(e)) => match e.local_name().as_ref() {
b"v" | b"t" => in_value = false,
b"c" => {
if let (Some((column, cell_type)), Some(current_row)) =
(cell.take(), row.as_mut())
{
let text = if cell_type == "s" {
let index = value.trim().parse::<usize>().map_err(|e| {
xml_error("worksheet", format!("bad shared string index: {e}"))
})?;
shared_strings.get(index).cloned().ok_or_else(|| {
xml_error(
"worksheet",
format!("shared string {index} out of range"),
)
})?
} else {
value.clone()
};
current_row.cells.insert(column, text);
}
}
b"row" => {
if let Some(current_row) = row.take() {
rows.push(current_row);
}
}
_ => {}
},
Ok(Event::Eof) => break,
Err(e) => return Err(xml_error("worksheet", e)),
_ => {}
}
}
Ok(rows)
}
fn cell_ref_and_type(element: &BytesStart<'_>) -> Result<(String, String), IdxError> {
let mut column = String::new();
let mut cell_type = String::new();
for attribute in element.attributes() {
let attribute = attribute.map_err(|e| xml_error("worksheet", e))?;
let raw = String::from_utf8_lossy(&attribute.value).to_string();
match attribute.key.local_name().as_ref() {
b"r" => column = raw.chars().take_while(char::is_ascii_alphabetic).collect(),
b"t" => cell_type = raw,
_ => {}
}
}
if column.is_empty() {
return Err(xml_error("worksheet", "cell without a reference"));
}
Ok((column, cell_type))
}
fn drafts_from_rows(rows: &[SheetRow]) -> Result<Vec<KseiHoldingDraft>, IdxError> {
let (header_index, columns) = rows
.iter()
.take(MAX_HEADER_SEARCH_ROWS)
.enumerate()
.find_map(|(index, row)| header_columns(row).map(|columns| (index, columns)))
.ok_or_else(|| {
IdxError::ParseError(
"ownership XLSX header did not match the above-1% holder register layout \
(DATE, SHARE_CODE, ... PERCENTAGE); the above-5% and investor-type \
workbooks are not supported"
.to_string(),
)
})?;
let mut drafts = Vec::new();
for row in &rows[header_index + 1..] {
let line = row.number;
let field = |name: &str| -> &str {
columns
.get(name)
.and_then(|column| row.cells.get(column))
.map(|value| value.trim())
.unwrap_or("")
};
let ticker_code = field("SHARE_CODE");
let investor_name = field("INVESTOR_NAME");
if ticker_code.is_empty() && investor_name.is_empty() {
continue;
}
if ticker_code.is_empty() || investor_name.is_empty() {
return Err(IdxError::ParseError(format!(
"ownership XLSX row {line} is missing SHARE_CODE or INVESTOR_NAME"
)));
}
drafts.push(KseiHoldingDraft {
ticker_code: ticker_code.to_string(),
issuer_name: optional(field("ISSUER_NAME")),
raw_investor_name: investor_name.to_string(),
investor_type: normalize_investor_type(field("INVESTOR_CLASSIFICATION")),
locality: normalize_locality(field("LOCAL_FOREIGN")),
nationality: optional(field("NATIONALITY")),
domicile: optional(field("DOMICILE")),
holdings_scripless: parse_shares(field("HOLDINGS_SCRIPLESS"), line, false)?,
holdings_scrip: parse_shares(field("HOLDINGS_SCRIP"), line, false)?,
total_shares: parse_shares(field("TOTAL_HOLDING_SHARES"), line, true)?,
percentage_bps: parse_percentage_bps(field("PERCENTAGE"), line)?,
report_date: parse_report_date(field("DATE"), line)?,
});
}
if drafts.is_empty() {
return Err(IdxError::ParseError(
"ownership XLSX contained a header but no holder rows".to_string(),
));
}
Ok(drafts)
}
/// Map header labels to column letters if `row` is exactly the expected header.
fn header_columns(row: &SheetRow) -> Option<HashMap<&'static str, String>> {
let mut found: HashMap<&'static str, String> = HashMap::new();
for (column, text) in &row.cells {
let mut label = text.trim().to_ascii_uppercase().replace(' ', "_");
if label.is_empty() {
continue;
}
// Feb/Mar 2026 workbooks name the column INVESTOR_TYPE and hold codes
// (CP, ID, ...); later ones use INVESTOR_CLASSIFICATION with words.
if label == "INVESTOR_TYPE" {
label = "INVESTOR_CLASSIFICATION".to_string();
}
let expected = EXPECTED_HEADER.iter().find(|name| **name == label)?;
found.insert(expected, column.clone());
}
(found.len() == EXPECTED_HEADER.len()).then_some(found)
}
fn optional(value: &str) -> Option<String> {
(!value.is_empty()).then(|| value.to_string())
}
fn parse_number(raw: &str, line: usize, name: &str) -> Result<f64, IdxError> {
raw.replace(',', "")
.parse::<f64>()
.ok()
.filter(|value| value.is_finite())
.ok_or_else(|| {
IdxError::ParseError(format!("ownership XLSX row {line}: invalid {name} '{raw}'"))
})
}
fn parse_shares(raw: &str, line: usize, required: bool) -> Result<i64, IdxError> {
if raw.is_empty() {
return if required {
Err(IdxError::ParseError(format!(
"ownership XLSX row {line}: missing TOTAL_HOLDING_SHARES"
)))
} else {
Ok(0)
};
}
let value = parse_number(raw, line, "share count")?;
if value < 0.0 || value.fract() != 0.0 || value > i64::MAX as f64 {
return Err(IdxError::ParseError(format!(
"ownership XLSX row {line}: share count '{raw}' is not a whole number"
)));
}
Ok(value as i64)
}
/// `54.94` (percent) -> 5494 basis points. Excel stores values such as
/// `1.1499999999999999`, so round to the nearest basis point.
fn parse_percentage_bps(raw: &str, line: usize) -> Result<i64, IdxError> {
let value = parse_number(raw, line, "PERCENTAGE")?;
if !(0.0..=100.0).contains(&value) {
return Err(IdxError::ParseError(format!(
"ownership XLSX row {line}: PERCENTAGE '{raw}' is out of range"
)));
}
Ok((value * 100.0).round() as i64)
}
/// DATE is an Excel serial day number (`46265` = 2026-08-31); accept the
/// KSEI text form (`31-Aug-2026`) as well in case the cell is stored as text.
fn parse_report_date(raw: &str, line: usize) -> Result<NaiveDate, IdxError> {
if let Ok(serial) = raw.parse::<f64>() {
let days = serial.trunc() as i64;
let epoch = NaiveDate::from_ymd_opt(1899, 12, 30).expect("valid Excel epoch");
if (1..=2_958_465).contains(&days) {
return Ok(epoch + Duration::days(days));
}
}
parse_ksei_date(raw).map_err(|_| {
IdxError::ParseError(format!("ownership XLSX row {line}: invalid DATE '{raw}'"))
})
}
#[cfg(test)]
mod tests {
use super::*;
fn fixture() -> Vec<KseiHoldingDraft> {
parse_above1_xlsx_bytes(include_bytes!(
"../../tests/fixtures/ksei_above1_20260831_excerpt.xlsx"
))
.expect("above-1% XLSX excerpt parses")
}
#[test]
fn parses_above1_xlsx_excerpt() {
let drafts = fixture();
// Pinned so a regression that silently drops rows fails loudly.
assert_eq!(drafts.len(), 17);
assert!(
drafts
.iter()
.all(|d| d.report_date == NaiveDate::from_ymd_opt(2026, 8, 31).unwrap())
);
let dwimuria = drafts
.iter()
.find(|d| {
d.ticker_code == "BBCA" && d.raw_investor_name == "PT DWIMURIA INVESTAMA ANDALAN"
})
.expect("BBCA controlling holder present");
assert_eq!(dwimuria.percentage_bps, 5494);
assert_eq!(
dwimuria.investor_type.as_ref().map(|t| t.0.as_str()),
Some("CORPORATE")
);
assert_eq!(
dwimuria.total_shares,
dwimuria.holdings_scripless + dwimuria.holdings_scrip
);
}
fn header_row(labels: &[&str]) -> SheetRow {
sheet_row(1, labels)
}
fn sheet_row(number: usize, values: &[&str]) -> SheetRow {
SheetRow {
number,
cells: values
.iter()
.enumerate()
.filter(|(_, value)| !value.is_empty())
.map(|(index, value)| {
(
((b'A' + index as u8) as char).to_string(),
value.to_string(),
)
})
.collect(),
}
}
#[test]
fn accepts_legacy_investor_type_header() {
let mut labels = EXPECTED_HEADER.to_vec();
labels[4] = "INVESTOR_TYPE";
let columns = header_columns(&header_row(&labels)).expect("legacy header accepted");
assert_eq!(columns["INVESTOR_CLASSIFICATION"], "E");
let row = sheet_row(
2,
&[
"46112",
"AADI",
"ADARO ANDALAN INDONESIA Tbk",
"GARIBALDI THOHIR",
"ID",
"L ",
"INDONESIAN",
"INDONESIA",
"454011607",
"0",
"454011607",
"5.83",
],
);
let drafts = drafts_from_rows(&[header_row(&labels), row]).expect("legacy rows parse");
assert_eq!(
drafts[0].investor_type.as_ref().map(|t| t.0.as_str()),
Some("ID")
);
assert_eq!(
drafts[0].report_date,
NaiveDate::from_ymd_opt(2026, 3, 31).unwrap()
);
assert!(drafts[0].locality.is_some());
}
#[test]
fn row_errors_report_excel_row_numbers() {
let missing_investor = sheet_row(
709,
&["46265", "BBCA", "BANK CENTRAL ASIA Tbk", "", "Corporate"],
);
let err = drafts_from_rows(&[header_row(&EXPECTED_HEADER), missing_investor])
.expect_err("row without investor rejected");
assert!(err.to_string().contains("row 709"), "{err}");
}
#[test]
fn rejects_header_with_unknown_or_missing_columns() {
let mut labels = EXPECTED_HEADER.to_vec();
labels[11] = "PERUBAHAN";
assert!(header_columns(&header_row(&labels)).is_none());
assert!(header_columns(&header_row(&EXPECTED_HEADER[..11])).is_none());
}
#[test]
fn rounds_float_noise_in_percentages() {
assert_eq!(parse_percentage_bps("1.1499999999999999", 1).unwrap(), 115);
assert_eq!(parse_percentage_bps("41.1", 1).unwrap(), 4110);
assert!(parse_percentage_bps("100.5", 1).is_err());
}
#[test]
fn parses_excel_serial_and_text_dates() {
let expected = NaiveDate::from_ymd_opt(2026, 8, 31).unwrap();
assert_eq!(parse_report_date("46265", 1).unwrap(), expected);
assert_eq!(parse_report_date("31-Aug-2026", 1).unwrap(), expected);
assert!(parse_report_date("not a date", 1).is_err());
}
#[test]
fn rejects_above5_workbook() {
let err = parse_above1_xlsx_bytes(include_bytes!(
"../../tests/fixtures/ksei_above5_20260923_excerpt.xlsx"
))
.expect_err("above-5% workbook must be rejected");
assert!(
err.to_string().contains("above-1% holder register layout"),
"{err}"
);
}
#[test]
fn rejects_non_xlsx_bytes() {
assert!(parse_above1_xlsx_bytes(b"%PDF-1.7 not a workbook").is_err());
}
}