futu_cache/qot_cache/
rt.rs

1use std::sync::Arc;
2
3use futu_core::qot_stock_key::QotSecurityKey;
4use tokio::sync::watch;
5
6use super::{CachedTimeShare, QotCache};
7
8#[derive(Debug, Clone, Copy, Eq, PartialEq)]
9pub enum RtAverageMode {
10    PriceCount,
11    TurnoverVolume,
12}
13
14#[derive(Debug, Clone)]
15pub enum RtPushApplyOutcome {
16    Updated(CachedTimeShare),
17    NotUpdate,
18    BlankInMiddle,
19    NotFound,
20}
21
22#[derive(Debug, Clone, Copy, Eq, PartialEq)]
23pub struct RtPullGeneration {
24    pub subscription_generation: u64,
25    pub backend_generation: u64,
26}
27
28pub struct RtPullFlightGuard {
29    cache: Arc<QotCache>,
30    cache_key: String,
31    completion: watch::Sender<bool>,
32}
33
34impl Drop for RtPullFlightGuard {
35    fn drop(&mut self) {
36        self.cache.rt_pull_in_flight.remove(&self.cache_key);
37        self.completion.send_replace(true);
38    }
39}
40
41impl QotCache {
42    pub fn make_rt_key(sec_key: &str, session: i32) -> String {
43        format!("{sec_key}:s{session}")
44    }
45
46    pub fn make_rt_key_broker(key: &QotSecurityKey, session: i32) -> String {
47        format!("{}:s{}", key.cache_key(), session)
48    }
49
50    pub fn begin_rt_pull(self: &Arc<Self>, key: &QotSecurityKey) -> Option<RtPullFlightGuard> {
51        let cache_key = key.cache_key();
52        match self.rt_pull_in_flight.entry(cache_key.clone()) {
53            dashmap::mapref::entry::Entry::Occupied(_) => None,
54            dashmap::mapref::entry::Entry::Vacant(entry) => {
55                let (completion, _) = watch::channel(false);
56                entry.insert(completion.clone());
57                Some(RtPullFlightGuard {
58                    cache: Arc::clone(self),
59                    cache_key,
60                    completion,
61                })
62            }
63        }
64    }
65
66    pub async fn wait_for_rt_pull(&self, key: &QotSecurityKey) {
67        let completion = self
68            .rt_pull_in_flight
69            .get(&key.cache_key())
70            .map(|entry| entry.value().clone());
71        if let Some(completion) = completion {
72            let mut receiver = completion.subscribe();
73            if !*receiver.borrow() {
74                let _ = receiver.changed().await;
75            }
76        }
77    }
78
79    pub fn update_rt_data_broker(
80        &self,
81        key: &QotSecurityKey,
82        session: i32,
83        rt_data: Vec<CachedTimeShare>,
84    ) {
85        let _publish = self.rt_data_publish_lock.write();
86        let cache_key = Self::make_rt_key_broker(key, session);
87        self.rt_data.insert(cache_key, rt_data);
88    }
89
90    pub fn publish_rt_pull_generation(
91        &self,
92        key: &QotSecurityKey,
93        generation: RtPullGeneration,
94        buckets: Vec<(i32, Vec<CachedTimeShare>)>,
95    ) -> bool {
96        let _publish = self.rt_data_publish_lock.write();
97        let generation_key = key.cache_key();
98        if self
99            .rt_pull_generations
100            .get(&generation_key)
101            .is_some_and(|current| current.backend_generation > generation.backend_generation)
102        {
103            return false;
104        }
105        for session in [0, 2, 3, 5] {
106            self.rt_data.remove(&Self::make_rt_key_broker(key, session));
107        }
108        for (session, points) in buckets {
109            if !points.is_empty() {
110                self.rt_data
111                    .insert(Self::make_rt_key_broker(key, session), points);
112            }
113        }
114        self.rt_pull_generations.insert(generation_key, generation);
115        true
116    }
117
118    pub fn get_rt_data_broker(
119        &self,
120        key: &QotSecurityKey,
121        session: i32,
122    ) -> Option<Vec<CachedTimeShare>> {
123        let _publish = self.rt_data_publish_lock.read();
124        let cache_key = Self::make_rt_key_broker(key, session);
125        self.rt_data.get(&cache_key).map(|value| value.clone())
126    }
127
128    pub fn apply_rt_push_point_broker(
129        &self,
130        key: &QotSecurityKey,
131        session: i32,
132        mut incoming: CachedTimeShare,
133        average_mode: RtAverageMode,
134    ) -> RtPushApplyOutcome {
135        let _publish = self.rt_data_publish_lock.write();
136        if incoming.is_blank {
137            return RtPushApplyOutcome::NotUpdate;
138        }
139        let cache_key = Self::make_rt_key_broker(key, session);
140        let Some(mut bucket) = self.rt_data.get_mut(&cache_key) else {
141            return RtPushApplyOutcome::NotFound;
142        };
143        let Some(index) = bucket
144            .iter()
145            .position(|point| point.timestamp == incoming.timestamp)
146        else {
147            return RtPushApplyOutcome::NotFound;
148        };
149        if bucket[..index].iter().any(|point| point.is_blank) {
150            return RtPushApplyOutcome::BlankInMiddle;
151        }
152        if rt_push_content_eq(&bucket[index], &incoming) {
153            return RtPushApplyOutcome::NotUpdate;
154        }
155        if bucket.get(index + 1).is_some_and(|point| !point.is_blank) {
156            return RtPushApplyOutcome::NotUpdate;
157        }
158
159        incoming.last_close_price = bucket[index].last_close_price;
160        bucket[index] = incoming;
161        recompute_rt_average(&mut bucket, average_mode);
162        RtPushApplyOutcome::Updated(bucket[index].clone())
163    }
164}
165
166fn rt_push_content_eq(left: &CachedTimeShare, right: &CachedTimeShare) -> bool {
167    left.time == right.time
168        && left.minute == right.minute
169        && left.is_blank == right.is_blank
170        && left.price == right.price
171        && left.volume == right.volume
172        && left.hp_volume == right.hp_volume
173        && left.turnover == right.turnover
174        && left.timestamp == right.timestamp
175}
176
177fn recompute_rt_average(points: &mut [CachedTimeShare], mode: RtAverageMode) {
178    let Some(start) = points.iter().position(|point| !point.is_blank) else {
179        return;
180    };
181    let mut total_price = 0.0;
182    let mut price_count = 0_u64;
183    let mut total_turnover = 0.0;
184    let mut total_volume = 0.0;
185    let mut last_average = None;
186    for point in points.iter_mut().skip(start) {
187        if point.is_blank {
188            break;
189        }
190        point.avg_price = match mode {
191            RtAverageMode::PriceCount => {
192                total_price += point.price;
193                price_count += 1;
194                total_price / price_count as f64
195            }
196            RtAverageMode::TurnoverVolume => {
197                total_turnover += point.turnover;
198                total_volume += point.hp_volume;
199                if total_volume > 0.0 {
200                    total_turnover / total_volume
201                } else {
202                    last_average.unwrap_or(point.price)
203                }
204            }
205        };
206        last_average = Some(point.avg_price);
207    }
208}
futu_cache/qot_cache/rt.rs

futu_cache/qot_cache/
rt.rs
