futu_cache/qot_cache/
kline.rs

1use futu_core::qot_stock_key::QotSecurityKey;
2use futu_domain_qot_klrt::KlineAggregateSession;
3
4use super::QotCache;
5
6/// K 线缓存
7#[derive(Debug, Clone, PartialEq)]
8pub struct CachedKLine {
9    pub time: String,
10    pub is_blank: bool,
11    pub open_price: f64,
12    pub high_price: f64,
13    pub low_price: f64,
14    pub close_price: f64,
15    pub last_close_price: f64,
16    pub volume: i64,
17    pub hp_volume: f64,
18    pub turnover: f64,
19    pub turnover_rate: f64,
20    pub pe: f64,
21    pub timestamp: f64,
22    /// Backend `KlineItem.point_type == 2`; `None` means the field was absent.
23    pub is_replenish: Option<bool>,
24    /// Backend event-contract direction (0=None, 1=Yes, 2=No).
25    /// `None` means the wire field was absent; generic KLine remains unaffected.
26    pub direction: Option<i32>,
27}
28
29/// K-line cache dimensions that must travel together.
30///
31/// Keeps the C++ cache dimensions `(rehab, kl_type, session)` as one typed value
32/// so call sites cannot accidentally swap positional `i32` arguments.
33#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
34pub struct KlineDims {
35    pub rehab: i32,
36    pub kl_type: i32,
37    pub session: KlineAggregateSession,
38}
39
40impl KlineDims {
41    #[must_use]
42    pub const fn new(rehab: i32, kl_type: i32, session: KlineAggregateSession) -> Self {
43        Self {
44            rehab,
45            kl_type,
46            session,
47        }
48    }
49}
50
51impl QotCache {
52    const KLINE_PUSH_CACHE_MAX_POINTS: usize = 2000;
53    const EVENT_CONTRACT_KLINE_MAX_POINTS_PER_DIRECTION: usize = 1000;
54
55    /// 构造 K 线 cache key (v1.4.106 codex 1140 F3 4-tuple).
56    ///
57    /// 之前 key 仅 `(sec_key, kl_type)` 2-tuple, 同股票同 KLType 但前复权 vs
58    /// 后复权 / RTH vs ETH 数据互相覆盖. 对齐 C++ APIServer_Qot_KL.cpp:
59    /// `GetNewestKLByCount(stock_id, enRehabType, enKLType, num, session, ...)`
60    /// 用 4 维 key.
61    ///
62    /// - `rehab`: proto Qot_Common.RehabType (0=None, 1=Forward, 2=Backward),
63    ///   对齐 backend `FTCmdKline.ExrightType`. 同一股票同一 kl_type 不同 rehab
64    ///   走独立 cache, 不互相覆盖.
65    /// - `kl_type`: proto Qot_Common.KLType (1=1Min, 2=Day, ..., 11=Quarter).
66    /// - `session`: typed C++ aggregate view (`Rth`, `Eth`, `All`), never a
67    ///   raw `FTCmdKline.RequestSection`. Push first folds the wire-order
68    ///   section list, then updates the applicable aggregate views atomically;
69    ///   pull/read select the exact connection aggregate.
70    pub fn make_kline_key_by_dims(sec_key: &str, dims: KlineDims) -> String {
71        format!(
72            "{sec_key}:r{}:k{}:s{}",
73            dims.rehab,
74            dims.kl_type,
75            dims.session.key_code()
76        )
77    }
78
79    pub fn make_kline_key(
80        sec_key: &str,
81        rehab: i32,
82        kl_type: i32,
83        session: KlineAggregateSession,
84    ) -> String {
85        Self::make_kline_key_by_dims(sec_key, KlineDims::new(rehab, kl_type, session))
86    }
87
88    /// Replace one exact KLine aggregate generation under the shared owner.
89    pub fn update_klines_by_dims(&self, sec_key: &str, dims: KlineDims, klines: Vec<CachedKLine>) {
90        let _owner = self.kline_data_lock.write();
91        self.update_klines_by_dims_unlocked(sec_key, dims, klines);
92    }
93
94    fn update_klines_by_dims_unlocked(
95        &self,
96        sec_key: &str,
97        dims: KlineDims,
98        mut klines: Vec<CachedKLine>,
99    ) {
100        fill_kline_last_close_like_cpp(&mut klines);
101        let cache_key = Self::make_kline_key_by_dims(sec_key, dims);
102        self.klines.insert(cache_key, klines);
103    }
104
105    pub fn update_klines(
106        &self,
107        sec_key: &str,
108        rehab: i32,
109        kl_type: i32,
110        session: KlineAggregateSession,
111        klines: Vec<CachedKLine>,
112    ) {
113        self.update_klines_by_dims(sec_key, KlineDims::new(rehab, kl_type, session), klines);
114    }
115
116    /// Merge one ordinary KLine push point with the existing pull generation.
117    ///
118    /// Ref: C++ `NNDataCenter/Quote/NNData_Qot_KLRT.cpp:790-895`. A matching timestamp is
119    /// replaced only when high-precision volume is not lower; the prior
120    /// `last_close_price` remains authoritative and the following point is
121    /// relinked. Only a point newer than the current tail may append.
122    pub fn upsert_kline_push_point(
123        &self,
124        sec_key: &str,
125        dims: KlineDims,
126        incoming: CachedKLine,
127    ) -> Option<CachedKLine> {
128        let _owner = self.kline_data_lock.write();
129        self.upsert_kline_push_point_unlocked(sec_key, dims, incoming)
130    }
131
132    fn upsert_kline_push_point_unlocked(
133        &self,
134        sec_key: &str,
135        dims: KlineDims,
136        mut incoming: CachedKLine,
137    ) -> Option<CachedKLine> {
138        let cache_key = Self::make_kline_key_by_dims(sec_key, dims);
139        let mut bucket = self.klines.entry(cache_key).or_default();
140
141        if let Some(index) = bucket
142            .iter()
143            .position(|point| point.timestamp == incoming.timestamp)
144        {
145            if incoming.hp_volume < bucket[index].hp_volume
146                || ordinary_kline_content_eq_like_cpp(&incoming, &bucket[index])
147            {
148                return None;
149            }
150            incoming.last_close_price = bucket[index].last_close_price;
151            let close_price = incoming.close_price;
152            bucket[index] = incoming;
153            if let Some(next) = bucket.get_mut(index + 1) {
154                next.last_close_price = close_price;
155            }
156            return Some(bucket[index].clone());
157        }
158
159        if let Some(last) = bucket.last() {
160            if incoming.timestamp <= last.timestamp {
161                return None;
162            }
163            incoming.last_close_price = last.close_price;
164        }
165        bucket.push(incoming);
166        if bucket.len() > Self::KLINE_PUSH_CACHE_MAX_POINTS {
167            let drain = bucket.len() - Self::KLINE_PUSH_CACHE_MAX_POINTS;
168            bucket.drain(0..drain);
169        }
170        bucket.last().cloned()
171    }
172
173    /// Apply one ordinary push point to every C++ aggregate cache view under a
174    /// single owner and return only the final aggregate's updated point.
175    ///
176    /// Ref: `NNData_Qot_KLRT.cpp:561-584`; the shared `pResult` is overwritten
177    /// in RTH -> ETH -> ALL order and only that final result reaches PushKL.
178    pub fn upsert_kline_push_aggregates(
179        &self,
180        sec_key: &str,
181        rehab: i32,
182        kl_type: i32,
183        sessions: &[KlineAggregateSession],
184        incoming: CachedKLine,
185    ) -> Option<CachedKLine> {
186        let _owner = self.kline_data_lock.write();
187        let mut final_result = None;
188        for session in sessions {
189            final_result = self.upsert_kline_push_point_unlocked(
190                sec_key,
191                KlineDims::new(rehab, kl_type, *session),
192                incoming.clone(),
193            );
194        }
195        final_result
196    }
197
198    /// Read one exact KLine aggregate generation under the shared owner.
199    pub fn get_klines_by_dims(&self, sec_key: &str, dims: KlineDims) -> Option<Vec<CachedKLine>> {
200        let _owner = self.kline_data_lock.read();
201        let cache_key = Self::make_kline_key_by_dims(sec_key, dims);
202        self.klines.get(&cache_key).map(|v| v.clone())
203    }
204
205    pub fn get_klines(
206        &self,
207        sec_key: &str,
208        rehab: i32,
209        kl_type: i32,
210        session: KlineAggregateSession,
211    ) -> Option<Vec<CachedKLine>> {
212        self.get_klines_by_dims(sec_key, KlineDims::new(rehab, kl_type, session))
213    }
214
215    /// **v1.4.110 Phase 2 Slice 5**: 更新 K 线 (broker-aware).
216    ///
217    /// 用 `QotSecurityKey::cache_key()` 作 prefix, broker_id=None 时退化到原行为.
218    /// composite 维度仍是 4-tuple `(rehab, kl_type, session)`, broker_id 是第 5
219    /// 维通过 `QotSecurityKey` 注入到 prefix.
220    pub fn update_klines_broker_by_dims(
221        &self,
222        key: &QotSecurityKey,
223        dims: KlineDims,
224        klines: Vec<CachedKLine>,
225    ) {
226        let _owner = self.kline_data_lock.write();
227        self.update_klines_by_dims_unlocked(&key.cache_key(), dims, klines);
228    }
229
230    /// Merge a late CMD6161 pull generation with any live points already
231    /// committed while the request was in flight.
232    ///
233    /// Ref: `NNData_Qot_KLRT.cpp:896-973`. Equal timestamps choose the larger
234    /// high-precision volume; equal-volume points that differ only in
235    /// last-close keep the existing live point. The merged chain then fills
236    /// non-first zero last-close values from the preceding close.
237    pub fn merge_kline_pull_generation_broker_by_dims(
238        &self,
239        key: &QotSecurityKey,
240        dims: KlineDims,
241        mut incoming: Vec<CachedKLine>,
242    ) {
243        let _owner = self.kline_data_lock.write();
244        incoming.sort_by(|left, right| left.timestamp.total_cmp(&right.timestamp));
245        let cache_key = Self::make_kline_key_by_dims(&key.cache_key(), dims);
246        let existing = self
247            .klines
248            .get(&cache_key)
249            .map_or_else(Vec::new, |points| points.clone());
250        if existing.is_empty() {
251            fill_kline_last_close_like_cpp(&mut incoming);
252            self.klines.insert(cache_key, incoming);
253            return;
254        }
255
256        let mut merged = Vec::with_capacity(existing.len().max(incoming.len()));
257        let (mut old_index, mut new_index) = (0, 0);
258        while old_index < existing.len() && new_index < incoming.len() {
259            let old = &existing[old_index];
260            let new = &incoming[new_index];
261            match old.timestamp.total_cmp(&new.timestamp) {
262                std::cmp::Ordering::Less => {
263                    merged.push(old.clone());
264                    old_index += 1;
265                }
266                std::cmp::Ordering::Greater => {
267                    merged.push(new.clone());
268                    new_index += 1;
269                }
270                std::cmp::Ordering::Equal => {
271                    let selected = if old.hp_volume > new.hp_volume
272                        || (old.hp_volume == new.hp_volume
273                            && ordinary_kline_content_eq_except_last_close(old, new))
274                    {
275                        old
276                    } else {
277                        new
278                    };
279                    merged.push(selected.clone());
280                    old_index += 1;
281                    new_index += 1;
282                }
283            }
284        }
285        merged.extend_from_slice(&existing[old_index..]);
286        merged.extend_from_slice(&incoming[new_index..]);
287        fill_kline_last_close_like_cpp(&mut merged);
288        if merged.len() > Self::KLINE_PUSH_CACHE_MAX_POINTS {
289            let drain = merged.len() - Self::KLINE_PUSH_CACHE_MAX_POINTS;
290            merged.drain(0..drain);
291        }
292        self.klines.insert(cache_key, merged);
293    }
294
295    /// Atomically replace one EventContract direction while retaining the
296    /// other directions in the shared generic K-line dimension bucket.
297    ///
298    /// Ref: frozen C++ aec0f6cda1
299    /// `NNProtoCenter/Quote/NNBiz_Qot_KLRT.cpp:793-805,896-908`.
300    /// EventContract direction is point metadata, not a generic cache-key
301    /// dimension. The DashMap entry guard makes concurrent YES/NO cold pulls
302    /// one read-modify-write transaction.
303    pub fn merge_event_contract_klines_broker_by_direction(
304        &self,
305        key: &QotSecurityKey,
306        dims: KlineDims,
307        direction: i32,
308        klines: Vec<CachedKLine>,
309    ) {
310        let _owner = self.kline_data_lock.write();
311        let cache_key = Self::make_kline_key_by_dims(&key.cache_key(), dims);
312        let mut bucket = self.klines.entry(cache_key).or_default();
313        bucket.retain(|point| point.direction.unwrap_or(0) != direction);
314        let mut replacement = klines;
315        replacement.sort_by(|left, right| left.timestamp.total_cmp(&right.timestamp));
316        if replacement.len() > Self::EVENT_CONTRACT_KLINE_MAX_POINTS_PER_DIRECTION {
317            let drain = replacement.len() - Self::EVENT_CONTRACT_KLINE_MAX_POINTS_PER_DIRECTION;
318            replacement.drain(0..drain);
319        }
320        bucket.extend(replacement);
321        bucket.sort_by(|left, right| {
322            left.timestamp.total_cmp(&right.timestamp).then_with(|| {
323                left.direction
324                    .unwrap_or(0)
325                    .cmp(&right.direction.unwrap_or(0))
326            })
327        });
328    }
329
330    /// Upsert one EventContract point inside its direction-scoped generation.
331    /// Same-timestamp updates replace directly (no ordinary volume gate),
332    /// older points insert in order, and retention is capped per direction.
333    ///
334    /// Ref: C++ `NNDataCenter/Quote/NNData_Qot_ECKline.cpp:27-64`.
335    pub fn upsert_event_contract_kline_push_broker_by_direction(
336        &self,
337        key: &QotSecurityKey,
338        dims: KlineDims,
339        direction: i32,
340        incoming: CachedKLine,
341    ) {
342        let _owner = self.kline_data_lock.write();
343        let cache_key = Self::make_kline_key_by_dims(&key.cache_key(), dims);
344        let mut bucket = self.klines.entry(cache_key).or_default();
345        let mut same_direction: Vec<CachedKLine> = bucket
346            .iter()
347            .filter(|point| point.direction.unwrap_or(0) == direction)
348            .cloned()
349            .collect();
350
351        match same_direction
352            .binary_search_by(|point| point.timestamp.total_cmp(&incoming.timestamp))
353        {
354            Ok(index) => same_direction[index] = incoming,
355            Err(index) => same_direction.insert(index, incoming),
356        }
357        if same_direction.len() > Self::EVENT_CONTRACT_KLINE_MAX_POINTS_PER_DIRECTION {
358            let drain = same_direction.len() - Self::EVENT_CONTRACT_KLINE_MAX_POINTS_PER_DIRECTION;
359            same_direction.drain(0..drain);
360        }
361
362        bucket.retain(|point| point.direction.unwrap_or(0) != direction);
363        bucket.extend(same_direction);
364        bucket.sort_by(|left, right| {
365            left.timestamp.total_cmp(&right.timestamp).then_with(|| {
366                left.direction
367                    .unwrap_or(0)
368                    .cmp(&right.direction.unwrap_or(0))
369            })
370        });
371    }
372
373    pub fn update_klines_broker(
374        &self,
375        key: &QotSecurityKey,
376        rehab: i32,
377        kl_type: i32,
378        session: KlineAggregateSession,
379        klines: Vec<CachedKLine>,
380    ) {
381        self.update_klines_broker_by_dims(key, KlineDims::new(rehab, kl_type, session), klines);
382    }
383
384    /// **v1.4.110 Phase 2 Slice 5**: 获取 K 线 (broker-aware).
385    pub fn get_klines_broker_by_dims(
386        &self,
387        key: &QotSecurityKey,
388        dims: KlineDims,
389    ) -> Option<Vec<CachedKLine>> {
390        let _owner = self.kline_data_lock.read();
391        let cache_key = Self::make_kline_key_by_dims(&key.cache_key(), dims);
392        self.klines.get(&cache_key).map(|v| v.clone())
393    }
394
395    pub fn get_klines_broker(
396        &self,
397        key: &QotSecurityKey,
398        rehab: i32,
399        kl_type: i32,
400        session: KlineAggregateSession,
401    ) -> Option<Vec<CachedKLine>> {
402        self.get_klines_broker_by_dims(key, KlineDims::new(rehab, kl_type, session))
403    }
404}
405
406fn ordinary_kline_content_eq_like_cpp(left: &CachedKLine, right: &CachedKLine) -> bool {
407    left.time == right.time
408        && left.is_blank == right.is_blank
409        && left.open_price == right.open_price
410        && left.high_price == right.high_price
411        && left.low_price == right.low_price
412        && left.close_price == right.close_price
413        && left.last_close_price == right.last_close_price
414        && left.hp_volume == right.hp_volume
415        && left.turnover == right.turnover
416        && left.turnover_rate == right.turnover_rate
417        && left.pe == right.pe
418        && left.timestamp == right.timestamp
419        && left.is_replenish == right.is_replenish
420        && left.direction.unwrap_or(0) == right.direction.unwrap_or(0)
421}
422
423fn ordinary_kline_content_eq_except_last_close(left: &CachedKLine, right: &CachedKLine) -> bool {
424    left.time == right.time
425        && left.is_blank == right.is_blank
426        && left.open_price == right.open_price
427        && left.high_price == right.high_price
428        && left.low_price == right.low_price
429        && left.close_price == right.close_price
430        && left.hp_volume == right.hp_volume
431        && left.turnover == right.turnover
432        && left.turnover_rate == right.turnover_rate
433        && left.pe == right.pe
434        && left.timestamp == right.timestamp
435}
436
437fn fill_kline_last_close_like_cpp(points: &mut [CachedKLine]) {
438    for index in 1..points.len() {
439        if points[index].last_close_price == 0.0 {
440            points[index].last_close_price = points[index - 1].close_price;
441        }
442    }
443}
futu_cache/qot_cache/kline.rs

futu_cache/qot_cache/
kline.rs
