futu_cache/qot_cache/
kline.rs

1use std::sync::Arc;
2
3use futu_core::qot_stock_key::QotSecurityKey;
4use futu_domain_qot_klrt::KlineAggregateSession;
5use tokio::sync::watch;
6
7use super::QotCache;
8
9/// Single-flight owner for one `(sec, rehab, kl_type, session)` CMD6161 pull.
10/// Dropping it releases the slot and wakes every `wait_for_kline_pull_*`.
11///
12/// Ref: C++ `NNBiz_Qot_KLRT::PullNewestKLByCount` returns the existing
13/// `SubInfo.nReqKey` while a pull is in flight and clears it on reply
14/// (`NNBiz_Qot_KLRT.cpp:1706-1717,735-748`).
15pub struct KlinePullFlightGuard {
16    cache: Arc<QotCache>,
17    cache_key: String,
18    completion: watch::Sender<bool>,
19}
20
21impl Drop for KlinePullFlightGuard {
22    fn drop(&mut self) {
23        self.cache.kline_pull_in_flight.remove(&self.cache_key);
24        self.completion.send_replace(true);
25    }
26}
27
28/// K 线缓存
29#[derive(Debug, Clone, PartialEq)]
30pub struct CachedKLine {
31    pub time: String,
32    pub is_blank: bool,
33    pub open_price: f64,
34    pub high_price: f64,
35    pub low_price: f64,
36    pub close_price: f64,
37    pub last_close_price: f64,
38    pub volume: i64,
39    pub hp_volume: f64,
40    pub turnover: f64,
41    pub turnover_rate: f64,
42    pub pe: f64,
43    pub timestamp: f64,
44    /// Backend `KlineItem.point_type == 2`; `None` means the field was absent.
45    pub is_replenish: Option<bool>,
46    /// Backend event-contract direction (0=None, 1=Yes, 2=No).
47    /// `None` means the wire field was absent; generic KLine remains unaffected.
48    pub direction: Option<i32>,
49}
50
51/// K-line cache dimensions that must travel together.
52///
53/// Keeps the C++ cache dimensions `(rehab, kl_type, session)` as one typed value
54/// so call sites cannot accidentally swap positional `i32` arguments.
55#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
56pub struct KlineDims {
57    pub rehab: i32,
58    pub kl_type: i32,
59    pub session: KlineAggregateSession,
60}
61
62impl KlineDims {
63    #[must_use]
64    pub const fn new(rehab: i32, kl_type: i32, session: KlineAggregateSession) -> Self {
65        Self {
66            rehab,
67            kl_type,
68            session,
69        }
70    }
71}
72
73impl QotCache {
74    const KLINE_PUSH_CACHE_MAX_POINTS: usize = 2000;
75    const EVENT_CONTRACT_KLINE_MAX_POINTS_PER_DIRECTION: usize = 1000;
76
77    /// 构造 K 线 cache key (v1.4.106 codex 1140 F3 4-tuple).
78    ///
79    /// 之前 key 仅 `(sec_key, kl_type)` 2-tuple, 同股票同 KLType 但前复权 vs
80    /// 后复权 / RTH vs ETH 数据互相覆盖. 对齐 C++ APIServer_Qot_KL.cpp:
81    /// `GetNewestKLByCount(stock_id, enRehabType, enKLType, num, session, ...)`
82    /// 用 4 维 key.
83    ///
84    /// - `rehab`: proto Qot_Common.RehabType (0=None, 1=Forward, 2=Backward),
85    ///   对齐 backend `FTCmdKline.ExrightType`. 同一股票同一 kl_type 不同 rehab
86    ///   走独立 cache, 不互相覆盖.
87    /// - `kl_type`: proto Qot_Common.KLType (1=1Min, 2=Day, ..., 11=Quarter).
88    /// - `session`: typed C++ aggregate view (`Rth`, `Eth`, `All`), never a
89    ///   raw `FTCmdKline.RequestSection`. Push first folds the wire-order
90    ///   section list, then updates the applicable aggregate views atomically;
91    ///   pull/read select the exact connection aggregate.
92    pub fn make_kline_key_by_dims(sec_key: &str, dims: KlineDims) -> String {
93        format!(
94            "{sec_key}:r{}:k{}:s{}",
95            dims.rehab,
96            dims.kl_type,
97            dims.session.key_code()
98        )
99    }
100
101    pub fn make_kline_key(
102        sec_key: &str,
103        rehab: i32,
104        kl_type: i32,
105        session: KlineAggregateSession,
106    ) -> String {
107        Self::make_kline_key_by_dims(sec_key, KlineDims::new(rehab, kl_type, session))
108    }
109
110    /// Replace one exact KLine aggregate generation under the shared owner.
111    pub fn update_klines_by_dims(&self, sec_key: &str, dims: KlineDims, klines: Vec<CachedKLine>) {
112        let _owner = self.kline_data_lock.write();
113        self.update_klines_by_dims_unlocked(sec_key, dims, klines);
114    }
115
116    fn update_klines_by_dims_unlocked(
117        &self,
118        sec_key: &str,
119        dims: KlineDims,
120        mut klines: Vec<CachedKLine>,
121    ) {
122        fill_kline_last_close_like_cpp(&mut klines);
123        let cache_key = Self::make_kline_key_by_dims(sec_key, dims);
124        self.klines.insert(cache_key, klines);
125    }
126
127    pub fn update_klines(
128        &self,
129        sec_key: &str,
130        rehab: i32,
131        kl_type: i32,
132        session: KlineAggregateSession,
133        klines: Vec<CachedKLine>,
134    ) {
135        self.update_klines_by_dims(sec_key, KlineDims::new(rehab, kl_type, session), klines);
136    }
137
138    /// Merge one ordinary KLine push point with the existing pull generation.
139    ///
140    /// Ref: C++ `NNDataCenter/Quote/NNData_Qot_KLRT.cpp:790-895`. A matching timestamp is
141    /// replaced only when high-precision volume is not lower; the prior
142    /// `last_close_price` remains authoritative and the following point is
143    /// relinked. Only a point newer than the current tail may append.
144    pub fn upsert_kline_push_point(
145        &self,
146        sec_key: &str,
147        dims: KlineDims,
148        incoming: CachedKLine,
149    ) -> Option<CachedKLine> {
150        let _owner = self.kline_data_lock.write();
151        self.upsert_kline_push_point_unlocked(sec_key, dims, incoming)
152    }
153
154    fn upsert_kline_push_point_unlocked(
155        &self,
156        sec_key: &str,
157        dims: KlineDims,
158        mut incoming: CachedKLine,
159    ) -> Option<CachedKLine> {
160        let cache_key = Self::make_kline_key_by_dims(sec_key, dims);
161        let mut bucket = self.klines.entry(cache_key).or_default();
162
163        if let Some(index) = bucket
164            .iter()
165            .position(|point| point.timestamp == incoming.timestamp)
166        {
167            if incoming.hp_volume < bucket[index].hp_volume
168                || ordinary_kline_content_eq_like_cpp(&incoming, &bucket[index])
169            {
170                return None;
171            }
172            incoming.last_close_price = bucket[index].last_close_price;
173            let close_price = incoming.close_price;
174            bucket[index] = incoming;
175            if let Some(next) = bucket.get_mut(index + 1) {
176                next.last_close_price = close_price;
177            }
178            return Some(bucket[index].clone());
179        }
180
181        if let Some(last) = bucket.last() {
182            if incoming.timestamp <= last.timestamp {
183                return None;
184            }
185            incoming.last_close_price = last.close_price;
186        }
187        bucket.push(incoming);
188        if bucket.len() > Self::KLINE_PUSH_CACHE_MAX_POINTS {
189            let drain = bucket.len() - Self::KLINE_PUSH_CACHE_MAX_POINTS;
190            bucket.drain(0..drain);
191        }
192        bucket.last().cloned()
193    }
194
195    /// Apply one ordinary push point to every C++ aggregate cache view under a
196    /// single owner and return only the final aggregate's updated point.
197    ///
198    /// Ref: `NNData_Qot_KLRT.cpp:561-584`; the shared `pResult` is overwritten
199    /// in RTH -> ETH -> ALL order and only that final result reaches PushKL.
200    pub fn upsert_kline_push_aggregates(
201        &self,
202        sec_key: &str,
203        rehab: i32,
204        kl_type: i32,
205        sessions: &[KlineAggregateSession],
206        incoming: CachedKLine,
207    ) -> Option<CachedKLine> {
208        let _owner = self.kline_data_lock.write();
209        let mut final_result = None;
210        for session in sessions {
211            final_result = self.upsert_kline_push_point_unlocked(
212                sec_key,
213                KlineDims::new(rehab, kl_type, *session),
214                incoming.clone(),
215            );
216        }
217        final_result
218    }
219
220    /// Read one exact KLine aggregate generation under the shared owner.
221    pub fn get_klines_by_dims(&self, sec_key: &str, dims: KlineDims) -> Option<Vec<CachedKLine>> {
222        let _owner = self.kline_data_lock.read();
223        let cache_key = Self::make_kline_key_by_dims(sec_key, dims);
224        self.klines.get(&cache_key).map(|v| v.clone())
225    }
226
227    pub fn get_klines(
228        &self,
229        sec_key: &str,
230        rehab: i32,
231        kl_type: i32,
232        session: KlineAggregateSession,
233    ) -> Option<Vec<CachedKLine>> {
234        self.get_klines_by_dims(sec_key, KlineDims::new(rehab, kl_type, session))
235    }
236
237    /// **v1.4.110 Phase 2 Slice 5**: 更新 K 线 (broker-aware).
238    ///
239    /// 用 `QotSecurityKey::cache_key()` 作 prefix, broker_id=None 时退化到原行为.
240    /// composite 维度仍是 4-tuple `(rehab, kl_type, session)`, broker_id 是第 5
241    /// 维通过 `QotSecurityKey` 注入到 prefix.
242    pub fn update_klines_broker_by_dims(
243        &self,
244        key: &QotSecurityKey,
245        dims: KlineDims,
246        klines: Vec<CachedKLine>,
247    ) {
248        let _owner = self.kline_data_lock.write();
249        self.update_klines_by_dims_unlocked(&key.cache_key(), dims, klines);
250    }
251
252    /// Merge a late CMD6161 pull generation with any live points already
253    /// committed while the request was in flight.
254    ///
255    /// Ref: `NNData_Qot_KLRT.cpp:896-973`. Equal timestamps choose the larger
256    /// high-precision volume; equal-volume points that differ only in
257    /// last-close keep the existing live point. The merged chain then fills
258    /// non-first zero last-close values from the preceding close.
259    pub fn merge_kline_pull_generation_broker_by_dims(
260        &self,
261        key: &QotSecurityKey,
262        dims: KlineDims,
263        mut incoming: Vec<CachedKLine>,
264    ) {
265        let _owner = self.kline_data_lock.write();
266        incoming.sort_by(|left, right| left.timestamp.total_cmp(&right.timestamp));
267        let cache_key = Self::make_kline_key_by_dims(&key.cache_key(), dims);
268        let existing = self
269            .klines
270            .get(&cache_key)
271            .map_or_else(Vec::new, |points| points.clone());
272        if existing.is_empty() {
273            fill_kline_last_close_like_cpp(&mut incoming);
274            self.klines.insert(cache_key, incoming);
275            return;
276        }
277
278        let mut merged = Vec::with_capacity(existing.len().max(incoming.len()));
279        let (mut old_index, mut new_index) = (0, 0);
280        while old_index < existing.len() && new_index < incoming.len() {
281            let old = &existing[old_index];
282            let new = &incoming[new_index];
283            match old.timestamp.total_cmp(&new.timestamp) {
284                std::cmp::Ordering::Less => {
285                    merged.push(old.clone());
286                    old_index += 1;
287                }
288                std::cmp::Ordering::Greater => {
289                    merged.push(new.clone());
290                    new_index += 1;
291                }
292                std::cmp::Ordering::Equal => {
293                    let selected = if old.hp_volume > new.hp_volume
294                        || (old.hp_volume == new.hp_volume
295                            && ordinary_kline_content_eq_except_last_close(old, new))
296                    {
297                        old
298                    } else {
299                        new
300                    };
301                    merged.push(selected.clone());
302                    old_index += 1;
303                    new_index += 1;
304                }
305            }
306        }
307        merged.extend_from_slice(&existing[old_index..]);
308        merged.extend_from_slice(&incoming[new_index..]);
309        fill_kline_last_close_like_cpp(&mut merged);
310        if merged.len() > Self::KLINE_PUSH_CACHE_MAX_POINTS {
311            let drain = merged.len() - Self::KLINE_PUSH_CACHE_MAX_POINTS;
312            merged.drain(0..drain);
313        }
314        self.klines.insert(cache_key, merged);
315    }
316
317    /// Atomically replace one EventContract direction while retaining the
318    /// other directions in the shared generic K-line dimension bucket.
319    ///
320    /// Ref: frozen C++ aec0f6cda1
321    /// `NNProtoCenter/Quote/NNBiz_Qot_KLRT.cpp:793-805,896-908`.
322    /// EventContract direction is point metadata, not a generic cache-key
323    /// dimension. The DashMap entry guard makes concurrent YES/NO cold pulls
324    /// one read-modify-write transaction.
325    pub fn merge_event_contract_klines_broker_by_direction(
326        &self,
327        key: &QotSecurityKey,
328        dims: KlineDims,
329        direction: i32,
330        klines: Vec<CachedKLine>,
331    ) {
332        let _owner = self.kline_data_lock.write();
333        let cache_key = Self::make_kline_key_by_dims(&key.cache_key(), dims);
334        let mut bucket = self.klines.entry(cache_key).or_default();
335        bucket.retain(|point| point.direction.unwrap_or(0) != direction);
336        let mut replacement = klines;
337        replacement.sort_by(|left, right| left.timestamp.total_cmp(&right.timestamp));
338        if replacement.len() > Self::EVENT_CONTRACT_KLINE_MAX_POINTS_PER_DIRECTION {
339            let drain = replacement.len() - Self::EVENT_CONTRACT_KLINE_MAX_POINTS_PER_DIRECTION;
340            replacement.drain(0..drain);
341        }
342        bucket.extend(replacement);
343        bucket.sort_by(|left, right| {
344            left.timestamp.total_cmp(&right.timestamp).then_with(|| {
345                left.direction
346                    .unwrap_or(0)
347                    .cmp(&right.direction.unwrap_or(0))
348            })
349        });
350    }
351
352    /// Upsert one EventContract point inside its direction-scoped generation.
353    /// Same-timestamp updates replace directly (no ordinary volume gate),
354    /// older points insert in order, and retention is capped per direction.
355    ///
356    /// Ref: C++ `NNDataCenter/Quote/NNData_Qot_ECKline.cpp:27-64`.
357    pub fn upsert_event_contract_kline_push_broker_by_direction(
358        &self,
359        key: &QotSecurityKey,
360        dims: KlineDims,
361        direction: i32,
362        incoming: CachedKLine,
363    ) {
364        let _owner = self.kline_data_lock.write();
365        let cache_key = Self::make_kline_key_by_dims(&key.cache_key(), dims);
366        let mut bucket = self.klines.entry(cache_key).or_default();
367        let mut same_direction: Vec<CachedKLine> = bucket
368            .iter()
369            .filter(|point| point.direction.unwrap_or(0) == direction)
370            .cloned()
371            .collect();
372
373        match same_direction
374            .binary_search_by(|point| point.timestamp.total_cmp(&incoming.timestamp))
375        {
376            Ok(index) => same_direction[index] = incoming,
377            Err(index) => same_direction.insert(index, incoming),
378        }
379        if same_direction.len() > Self::EVENT_CONTRACT_KLINE_MAX_POINTS_PER_DIRECTION {
380            let drain = same_direction.len() - Self::EVENT_CONTRACT_KLINE_MAX_POINTS_PER_DIRECTION;
381            same_direction.drain(0..drain);
382        }
383
384        bucket.retain(|point| point.direction.unwrap_or(0) != direction);
385        bucket.extend(same_direction);
386        bucket.sort_by(|left, right| {
387            left.timestamp.total_cmp(&right.timestamp).then_with(|| {
388                left.direction
389                    .unwrap_or(0)
390                    .cmp(&right.direction.unwrap_or(0))
391            })
392        });
393    }
394
395    pub fn update_klines_broker(
396        &self,
397        key: &QotSecurityKey,
398        rehab: i32,
399        kl_type: i32,
400        session: KlineAggregateSession,
401        klines: Vec<CachedKLine>,
402    ) {
403        self.update_klines_broker_by_dims(key, KlineDims::new(rehab, kl_type, session), klines);
404    }
405
406    /// **v1.4.110 Phase 2 Slice 5**: 获取 K 线 (broker-aware).
407    pub fn get_klines_broker_by_dims(
408        &self,
409        key: &QotSecurityKey,
410        dims: KlineDims,
411    ) -> Option<Vec<CachedKLine>> {
412        let _owner = self.kline_data_lock.read();
413        let cache_key = Self::make_kline_key_by_dims(&key.cache_key(), dims);
414        self.klines.get(&cache_key).map(|v| v.clone())
415    }
416
417    /// Return the newest non-blank KLine only when that exact point's previous
418    /// close is authoritative.
419    ///
420    /// C++ `GetNewestKlineByCount` removes trailing blank/future points before
421    /// selecting the newest public item (`NNData_Qot_KLRT.cpp:178-192`). This
422    /// accessor keeps that scan under the cache read owner and clones one point
423    /// instead of cloning the full bucket.
424    pub fn latest_authoritative_kline_broker_by_dims(
425        &self,
426        key: &QotSecurityKey,
427        dims: KlineDims,
428    ) -> Option<CachedKLine> {
429        let _owner = self.kline_data_lock.read();
430        let cache_key = Self::make_kline_key_by_dims(&key.cache_key(), dims);
431        self.klines.get(&cache_key).and_then(|points| {
432            let latest = points.iter().rev().find(|point| !point.is_blank)?;
433            (latest.last_close_price > 0.0).then(|| latest.clone())
434        })
435    }
436
437    pub fn get_klines_broker(
438        &self,
439        key: &QotSecurityKey,
440        rehab: i32,
441        kl_type: i32,
442        session: KlineAggregateSession,
443    ) -> Option<Vec<CachedKLine>> {
444        self.get_klines_broker_by_dims(key, KlineDims::new(rehab, kl_type, session))
445    }
446
447    /// C++ `INNData_Qot_KLRT::GetKlineCount`: points up to and including the
448    /// last non-blank point (trailing blank points are not counted).
449    ///
450    /// Ref: `NNData_Qot_KLRT.cpp:260-283`.
451    pub fn kline_count_like_cpp_broker_by_dims(
452        &self,
453        key: &QotSecurityKey,
454        dims: KlineDims,
455    ) -> usize {
456        let _owner = self.kline_data_lock.read();
457        let cache_key = Self::make_kline_key_by_dims(&key.cache_key(), dims);
458        self.klines.get(&cache_key).map_or(0, |points| {
459            points
460                .iter()
461                .rposition(|point| !point.is_blank)
462                .map_or(0, |index| index + 1)
463        })
464    }
465
466    /// C++ `INNData_Qot_KLRT::HasOlderKline`: missing cache entry or a bucket
467    /// that no pull generation has written yet reports `true`
468    /// (`Ndt_Qot_KlineCacheData` constructor default).
469    ///
470    /// Ref: `NNData_Qot_KLRT.cpp:285-309,777-782`.
471    pub fn kline_has_older_broker_by_dims(&self, key: &QotSecurityKey, dims: KlineDims) -> bool {
472        let cache_key = Self::make_kline_key_by_dims(&key.cache_key(), dims);
473        self.kline_has_older
474            .get(&cache_key)
475            .is_none_or(|flag| *flag)
476    }
477
478    /// Record `KlineRsp.has_older_item` for the pulled session bucket only.
479    ///
480    /// Ref: `NNData_Qot_KLRT.cpp:533-556` (`AddKlinePoints`, `KLRTSrc_Pull`
481    /// branch writes `bHasOlderData{,BA,All}` for the requested session).
482    pub fn set_kline_has_older_broker_by_dims(
483        &self,
484        key: &QotSecurityKey,
485        dims: KlineDims,
486        has_older: bool,
487    ) {
488        let cache_key = Self::make_kline_key_by_dims(&key.cache_key(), dims);
489        self.kline_has_older.insert(cache_key, has_older);
490    }
491
492    /// C++ `IKLRTCore::IsHasKLData`: the cache satisfies a `GetKL` request when
493    /// it already holds `req_num` points or the backend has no older points to
494    /// pull. Otherwise the caller must pull CMD6161 and merge before replying.
495    ///
496    /// Ref: `APIServer/Business/Quote/IKLRTCore.cpp:6-19`.
497    pub fn kline_has_data_like_cpp_broker_by_dims(
498        &self,
499        key: &QotSecurityKey,
500        dims: KlineDims,
501        req_num: usize,
502    ) -> bool {
503        if req_num == 0 {
504            return false;
505        }
506        self.kline_count_like_cpp_broker_by_dims(key, dims) >= req_num
507            || !self.kline_has_older_broker_by_dims(key, dims)
508    }
509
510    /// Claim the single in-flight CMD6161 slot for one `(sec, dims)`.
511    /// `None` means another pull for the same identity is in flight.
512    pub fn begin_kline_pull_broker_by_dims(
513        self: &Arc<Self>,
514        key: &QotSecurityKey,
515        dims: KlineDims,
516    ) -> Option<KlinePullFlightGuard> {
517        let cache_key = Self::make_kline_key_by_dims(&key.cache_key(), dims);
518        match self.kline_pull_in_flight.entry(cache_key.clone()) {
519            dashmap::mapref::entry::Entry::Occupied(_) => None,
520            dashmap::mapref::entry::Entry::Vacant(entry) => {
521                let (completion, _) = watch::channel(false);
522                entry.insert(completion.clone());
523                Some(KlinePullFlightGuard {
524                    cache: Arc::clone(self),
525                    cache_key,
526                    completion,
527                })
528            }
529        }
530    }
531
532    /// Wait until the in-flight CMD6161 pull for `(sec, dims)` (if any) has
533    /// completed. Returns immediately when nothing is in flight.
534    pub async fn wait_for_kline_pull_broker_by_dims(&self, key: &QotSecurityKey, dims: KlineDims) {
535        let cache_key = Self::make_kline_key_by_dims(&key.cache_key(), dims);
536        let completion = self
537            .kline_pull_in_flight
538            .get(&cache_key)
539            .map(|entry| entry.value().clone());
540        if let Some(completion) = completion {
541            let mut receiver = completion.subscribe();
542            if !*receiver.borrow() {
543                let _ = receiver.changed().await;
544            }
545        }
546    }
547}
548
549fn ordinary_kline_content_eq_like_cpp(left: &CachedKLine, right: &CachedKLine) -> bool {
550    left.time == right.time
551        && left.is_blank == right.is_blank
552        && left.open_price == right.open_price
553        && left.high_price == right.high_price
554        && left.low_price == right.low_price
555        && left.close_price == right.close_price
556        && left.last_close_price == right.last_close_price
557        && left.hp_volume == right.hp_volume
558        && left.turnover == right.turnover
559        && left.turnover_rate == right.turnover_rate
560        && left.pe == right.pe
561        && left.timestamp == right.timestamp
562        && left.is_replenish == right.is_replenish
563        && left.direction.unwrap_or(0) == right.direction.unwrap_or(0)
564}
565
566fn ordinary_kline_content_eq_except_last_close(left: &CachedKLine, right: &CachedKLine) -> bool {
567    left.time == right.time
568        && left.is_blank == right.is_blank
569        && left.open_price == right.open_price
570        && left.high_price == right.high_price
571        && left.low_price == right.low_price
572        && left.close_price == right.close_price
573        && left.hp_volume == right.hp_volume
574        && left.turnover == right.turnover
575        && left.turnover_rate == right.turnover_rate
576        && left.pe == right.pe
577        && left.timestamp == right.timestamp
578}
579
580fn fill_kline_last_close_like_cpp(points: &mut [CachedKLine]) {
581    for index in 1..points.len() {
582        if points[index].last_close_price == 0.0 {
583            points[index].last_close_price = points[index - 1].close_price;
584        }
585    }
586}
futu_cache/qot_cache/kline.rs

futu_cache/qot_cache/
kline.rs
