0xrsydn/idx-cli
1
mirror of https://github.com/0xrsydn/idx-cli.git synced 2026-11-07 09:45:36 +00:00
Code Issues Projects Releases 1 Packages Wiki Activity

fix(movements): read a query's pages as one answer and keep a partial one

Browse source 
- Each page was cached on its own, so a multi-page answer could join
  pages read at different times from a feed that gains rows at the top,
  repeating some rows and missing others. A query's pages are now read
  together and cached as one entry.
- A failed later page threw away the pages already read. They are now
  returned with incomplete saying why, and not cached.
- The library did not cap pages at MAX_PAGES itself; it clamps now.
- The hint said to raise --pages even at 10; at 10 it says to narrow the
  dates. -o jsonl, which streams rows only, gives the hint on stderr.
- A duplicate match arm in the CLI was dead; it is gone.

Claude-Session: https://claude.ai/code/session_01RqPPVeVDrnLoP7NUZmLRxY
This commit is contained in:
Rasyidan Akbar F. 2026-11-04 17:39:10 +07:00
commit 0c7320a343
11 changed files with 159 additions and 54 deletions
Download patch file Download diff file Expand all files Collapse all files

4
crates/idx-core/src/api/mock.rs
View file

@ -271,6 +271,10 @@ impl ShareholderMovementProvider for MockStockbitProvider {
page: u32,
) -> Result<MovementPage, IdxError> {
let subject = query.symbol.as_deref().unwrap_or("shareholder movements");
// `IDX_MOCK_MOVEMENTS_FAIL_PAGE=N` rate-limits page N alone.
if std::env::var("IDX_MOCK_MOVEMENTS_FAIL_PAGE").ok() == Some(page.to_string()) {
return Err(IdxError::RateLimited);
}
let mut name = "majorholder".to_string();
if let Some(symbol) = &query.symbol {
name.push('_');

127
crates/idx-core/src/movements/mod.rs
View file

@ -1,7 +1,9 @@
//! Shareholder movements from the user's Stockbit session: reported changes
//! in insiders' and major holders' stakes (IDX insider disclosures and KSEI's
//! 5% holder changes), one stock's or the market's, newest first. Each page
//! is 50 rows and one request (paced 1 s apart), cached for an hour.
//! is 50 rows and one request (paced 1 s apart). A query's pages are read
//! together and cached together for an hour, so they come from one moment
//! of a feed that gains rows at the top.
//!
//! These are the reports as filed. Whether they predict anything is not
//! claimed here: share plans read as a cluster of buys at one price, and a
@ -9,12 +11,15 @@
pub mod types;
use std::cell::RefCell;
use chrono::Utc;
use crate::api::{shareholder_movement_provider, stockbit};
use crate::api::{ShareholderMovementProvider, shareholder_movement_provider, stockbit};
use crate::client::{CachePolicy, CacheSpec, Fetched, Idx, cached_fetch};
use crate::error::IdxError;
use types::{MovementList, MovementQuery};
use crate::runtime;
use types::{MAX_PAGES, MovementList, MovementQuery};
/// The feed gains rows through the day; an hour keeps a session's repeated
/// queries cheap without hiding a day's filings for long.
@ -26,15 +31,19 @@ pub struct Movements<'a> {
}
impl Movements<'_> {
/// Up to `query.pages` pages of the feed, newest first; `more` says
/// whether rows remain past them. The result is as old as its oldest page
/// and stale if any page is.
/// Up to `query.pages` pages of the feed (at most `MAX_PAGES`), newest
/// first; `more` says whether rows remain past them.
///
/// When a page after the first fails and no cached answer is at hand,
/// the rows read before it are returned with `incomplete` saying why,
/// and are not cached.
pub fn list(
&self,
query: &MovementQuery,
policy: CachePolicy,
) -> Result<Fetched<MovementList>, IdxError> {
let mut query = query.clone();
query.pages = query.pages.clamp(1, MAX_PAGES);
if let Some(symbol) = &query.symbol {
let (resolved, _) = self.idx.idx_code(symbol, "shareholder movements")?;
query.symbol = Some(resolved);
@ -44,50 +53,80 @@ impl Movements<'_> {
.symbol
.clone()
.unwrap_or_else(|| "shareholder movements".to_string());
let mut list = Fetched {
data: MovementList::default(),
source: stockbit::SOURCE,
fetched_at: Utc::now(),
stale: false,
};
for page in 1..=query.pages.max(1) {
let key = cache_key(&query, page);
let part = cached_fetch(
&self.idx.cache,
CacheSpec {
bucket: BUCKET,
key: &key,
ttl_secs: TTL_SECS,
subject: &subject,
let key = cache_key(&query);
let partial = RefCell::new(None);
let result = cached_fetch(
&self.idx.cache,
CacheSpec {
bucket: BUCKET,
key: &key,
ttl_secs: TTL_SECS,
subject: &subject,
source: stockbit::SOURCE,
},
policy,
|| read_pages(provider.as_ref(), &query, &partial),
Ok,
);
match (result, partial.into_inner()) {
(Err(err), Some(mut list)) => {
runtime::warn(format!(
"{err}; showing the {} rows read before it",
list.movements.len()
));
list.incomplete = Some(err.to_string());
Ok(Fetched {
data: list,
source: stockbit::SOURCE,
},
policy,
|| provider.shareholder_movements(&query, page),
Ok,
)?;
list.fetched_at = list.fetched_at.min(part.fetched_at);
list.stale |= part.stale;
list.data.movements.extend(part.data.movements);
list.data.more = part.data.more;
if !part.data.more {
break;
fetched_at: Utc::now(),
stale: false,
})
}
(result, _) => result,
}
Ok(list)
}
}
/// Every filter and the page, so no two queries share an entry.
fn cache_key(query: &MovementQuery, page: u32) -> String {
/// Reads the query's pages in turn. A failure after the first page leaves
/// the rows read so far in `partial` and fails, so they are not cached.
fn read_pages(
provider: &dyn ShareholderMovementProvider,
query: &MovementQuery,
partial: &RefCell<Option<MovementList>>,
) -> Result<MovementList, IdxError> {
let mut list = MovementList::default();
for page in 1..=query.pages {
match provider.shareholder_movements(query, page) {
Ok(part) => {
list.movements.extend(part.movements);
list.more = part.more;
if !part.more {
break;
}
}
Err(err) if page > 1 => {
list.more = true;
*partial.borrow_mut() = Some(list);
return Err(err);
}
Err(err) => return Err(err),
}
}
Ok(list)
}
/// Every filter and the page count, so no two queries share an entry.
fn cache_key(query: &MovementQuery) -> String {
let opt = |value: Option<String>| value.unwrap_or_else(|| "-".to_string());
format!(
"{}:{}:{}:{}:{}:{}:p{page}",
"{}:{}:{}:{}:{}:{}:{}p",
query.symbol.as_deref().unwrap_or("market"),
opt(query.from.map(|d| d.to_string())),
opt(query.to.map(|d| d.to_string())),
query.holder_id.as_deref().unwrap_or("-"),
query.action.map(|a| a.as_str()).unwrap_or("-"),
query.source.map(|s| s.as_str()).unwrap_or("-"),
query.pages,
)
}
@ -99,25 +138,27 @@ mod tests {
#[test]
fn cache_keys_differ_by_every_filter() {
let base = MovementQuery::from_args(None, None, None, None, 1).unwrap();
let mut keys = vec![cache_key(&base, 1), cache_key(&base, 2)];
let mut two = base.clone();
two.pages = 2;
let mut keys = vec![cache_key(&base), cache_key(&two)];
let mut q = base.clone();
q.symbol = Some("BBCA.JK".to_string());
keys.push(cache_key(&q, 1));
keys.push(cache_key(&q));
let mut q = base.clone();
q.from = chrono::NaiveDate::from_ymd_opt(2026, 10, 1);
keys.push(cache_key(&q, 1));
keys.push(cache_key(&q));
let mut q = base.clone();
q.to = chrono::NaiveDate::from_ymd_opt(2026, 10, 1);
keys.push(cache_key(&q, 1));
keys.push(cache_key(&q));
let mut q = base.clone();
q.holder_id = Some("14074".to_string());
keys.push(cache_key(&q, 1));
keys.push(cache_key(&q));
let mut q = base.clone();
q.action = Some(ActionFilter::Sell);
keys.push(cache_key(&q, 1));
keys.push(cache_key(&q));
let mut q = base;
q.source = Some(MovementSource::Ksei);
keys.push(cache_key(&q, 1));
keys.push(cache_key(&q));
let unique: std::collections::HashSet<_> = keys.iter().collect();
assert_eq!(unique.len(), keys.len(), "{keys:?}");
}

5
crates/idx-core/src/movements/types.rs
View file

@ -288,6 +288,11 @@ pub struct MovementList {
/// More rows exist past the pages read (raise `--pages`, or narrow the
/// dates).
pub more: bool,
/// Why reading stopped before the pages asked for: a later page failed,
/// and `movements` holds the rows read before it. Absent when every page
/// asked for was read.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub incomplete: Option<String>,
}
#[cfg(test)]