longbridge/market/
context.rs

1use std::sync::Arc;
2
3use longbridge_httpcli::{DcRegion, HttpClient, Json, Method};
4use serde::{Serialize, de::DeserializeOwned};
5use tracing::{Subscriber, dispatcher, instrument::WithSubscriber};
6
7use crate::{
8    Config, Result,
9    market::types::*,
10    utils::counter::{counter_id_to_symbol, index_symbol_to_counter_id, symbol_to_counter_id},
11};
12
13/// Convert a Unix-seconds value (integer or string) to RFC 3339.
14fn unix_secs_to_rfc3339(ts: i64) -> String {
15    time::OffsetDateTime::from_unix_timestamp(ts)
16        .map(|dt| {
17            use time::format_description::well_known::Rfc3339;
18            dt.format(&Rfc3339).unwrap_or_default()
19        })
20        .unwrap_or_else(|_| ts.to_string())
21}
22
23/// Convert a Unix-seconds string to RFC 3339.
24fn unix_secs_str_to_rfc3339(s: &str) -> String {
25    s.parse::<i64>()
26        .map(unix_secs_to_rfc3339)
27        .unwrap_or_else(|_| s.to_string())
28}
29
30struct InnerMarketContext {
31    http_cli: HttpClient,
32    log_subscriber: Arc<dyn Subscriber + Send + Sync>,
33}
34
35impl Drop for InnerMarketContext {
36    fn drop(&mut self) {
37        dispatcher::with_default(&self.log_subscriber.clone().into(), || {
38            tracing::info!("market context dropped");
39        });
40    }
41}
42
43/// Market data context — broker holdings, A/H premium, trade statistics,
44/// market anomalies, index constituents and more.
45#[derive(Clone)]
46pub struct MarketContext(Arc<InnerMarketContext>);
47
48impl MarketContext {
49    /// Create a [`MarketContext`]
50    pub fn new(config: Arc<Config>) -> Self {
51        let log_subscriber = config.create_log_subscriber("market");
52        dispatcher::with_default(&log_subscriber.clone().into(), || {
53            tracing::info!(language = ?config.language, "creating market context");
54        });
55        let ctx = Self(Arc::new(InnerMarketContext {
56            http_cli: config.create_http_client(),
57            log_subscriber,
58        }));
59        dispatcher::with_default(&ctx.0.log_subscriber.clone().into(), || {
60            tracing::info!("market context created");
61        });
62        ctx
63    }
64
65    /// Returns the log subscriber
66    #[inline]
67    pub fn log_subscriber(&self) -> Arc<dyn Subscriber + Send + Sync> {
68        self.0.log_subscriber.clone()
69    }
70
71    async fn get<R, Q>(&self, path: &'static str, query: Q) -> Result<R>
72    where
73        R: DeserializeOwned + Send + Sync + 'static,
74        Q: Serialize + Send + Sync,
75    {
76        Ok(self
77            .0
78            .http_cli
79            .request(Method::GET, path)
80            .query_params(query)
81            .response::<Json<R>>()
82            .send()
83            .with_subscriber(self.0.log_subscriber.clone())
84            .await?
85            .0)
86    }
87
88    /// Like [`get`](Self::get), but restricted to a single data center. Used by
89    /// region-limited endpoints (e.g. AP-only broker holdings).
90    async fn get_dc<R, Q>(&self, path: &'static str, query: Q, dc_restrict: DcRegion) -> Result<R>
91    where
92        R: DeserializeOwned + Send + Sync + 'static,
93        Q: Serialize + Send + Sync,
94    {
95        Ok(self
96            .0
97            .http_cli
98            .request(Method::GET, path)
99            .dc_restrict(dc_restrict)
100            .query_params(query)
101            .response::<Json<R>>()
102            .send()
103            .with_subscriber(self.0.log_subscriber.clone())
104            .await?
105            .0)
106    }
107
108    async fn post<R, B>(&self, path: &'static str, body: B) -> Result<R>
109    where
110        R: DeserializeOwned + Send + Sync + 'static,
111        B: std::fmt::Debug + Serialize + Send + Sync + 'static,
112    {
113        Ok(self
114            .0
115            .http_cli
116            .request(Method::POST, path)
117            .body(Json(body))
118            .response::<Json<R>>()
119            .send()
120            .with_subscriber(self.0.log_subscriber.clone())
121            .await?
122            .0)
123    }
124
125    // ── market_status ─────────────────────────────────────────────
126
127    /// Get current trading status for all markets.
128    ///
129    /// Path: `GET /v1/quote/market-status`
130    pub async fn market_status(&self) -> Result<MarketStatusResponse> {
131        #[derive(Serialize)]
132        struct Empty {}
133        self.get("/v1/quote/market-status", Empty {}).await
134    }
135
136    // ── broker_holding ────────────────────────────────────────────
137
138    /// Get top broker holdings (buy/sell leaders) for a security.
139    ///
140    /// Path: `GET /v1/quote/broker-holding`
141    pub async fn broker_holding(
142        &self,
143        symbol: impl Into<String>,
144        period: BrokerHoldingPeriod,
145    ) -> Result<BrokerHoldingTop> {
146        let period_str = match period {
147            BrokerHoldingPeriod::Rct1 => "rct_1",
148            BrokerHoldingPeriod::Rct5 => "rct_5",
149            BrokerHoldingPeriod::Rct20 => "rct_20",
150            BrokerHoldingPeriod::Rct60 => "rct_60",
151        };
152        #[derive(Serialize)]
153        struct Query {
154            counter_id: String,
155            #[serde(rename = "type")]
156            period: &'static str,
157        }
158        self.get_dc(
159            "/v1/quote/broker-holding",
160            Query {
161                counter_id: symbol_to_counter_id(&symbol.into()),
162                period: period_str,
163            },
164            DcRegion::Ap,
165        )
166        .await
167    }
168
169    /// Get full broker holding details for a security.
170    ///
171    /// Path: `GET /v1/quote/broker-holding/detail`
172    pub async fn broker_holding_detail(
173        &self,
174        symbol: impl Into<String>,
175    ) -> Result<BrokerHoldingDetail> {
176        #[derive(Serialize)]
177        struct Query {
178            counter_id: String,
179        }
180        self.get_dc(
181            "/v1/quote/broker-holding/detail",
182            Query {
183                counter_id: symbol_to_counter_id(&symbol.into()),
184            },
185            DcRegion::Ap,
186        )
187        .await
188    }
189
190    /// Get daily holding history for a specific broker.
191    ///
192    /// Path: `GET /v1/quote/broker-holding/daily`
193    pub async fn broker_holding_daily(
194        &self,
195        symbol: impl Into<String>,
196        broker_id: impl Into<String>,
197    ) -> Result<BrokerHoldingDailyHistory> {
198        #[derive(Serialize)]
199        struct Query {
200            counter_id: String,
201            parti_number: String,
202        }
203        self.get_dc(
204            "/v1/quote/broker-holding/daily",
205            Query {
206                counter_id: symbol_to_counter_id(&symbol.into()),
207                parti_number: broker_id.into(),
208            },
209            DcRegion::Ap,
210        )
211        .await
212    }
213
214    // ── ah_premium ────────────────────────────────────────────────
215
216    /// Get A/H premium K-line data for a dual-listed security.
217    ///
218    /// Path: `GET /v1/quote/ahpremium/klines`
219    pub async fn ah_premium(
220        &self,
221        symbol: impl Into<String>,
222        period: AhPremiumPeriod,
223        count: u32,
224    ) -> Result<AhPremiumKlines> {
225        #[derive(Serialize)]
226        struct Query {
227            counter_id: String,
228            line_type: &'static str,
229            line_num: u32,
230        }
231        self.get(
232            "/v1/quote/ahpremium/klines",
233            Query {
234                counter_id: symbol_to_counter_id(&symbol.into()),
235                line_type: period.to_line_type(),
236                line_num: count,
237            },
238        )
239        .await
240    }
241
242    /// Get A/H premium intraday data for a dual-listed security.
243    ///
244    /// Path: `GET /v1/quote/ahpremium/timeshares`
245    pub async fn ah_premium_intraday(
246        &self,
247        symbol: impl Into<String>,
248    ) -> Result<AhPremiumIntraday> {
249        #[derive(Serialize)]
250        struct Query {
251            counter_id: String,
252            days: &'static str,
253        }
254        self.get(
255            "/v1/quote/ahpremium/timeshares",
256            Query {
257                counter_id: symbol_to_counter_id(&symbol.into()),
258                days: "1",
259            },
260        )
261        .await
262    }
263
264    // ── trade_stats ───────────────────────────────────────────────
265
266    /// Get buy/sell/neutral trade statistics for a security.
267    ///
268    /// Path: `GET /v1/quote/trades-statistics`
269    pub async fn trade_stats(&self, symbol: impl Into<String>) -> Result<TradeStatsResponse> {
270        #[derive(Serialize)]
271        struct Query {
272            counter_id: String,
273        }
274        self.get(
275            "/v1/quote/trades-statistics",
276            Query {
277                counter_id: symbol_to_counter_id(&symbol.into()),
278            },
279        )
280        .await
281    }
282
283    // ── anomaly ───────────────────────────────────────────────────
284
285    /// Get market anomaly alerts (unusual price/volume events).
286    ///
287    /// Path: `GET /v1/quote/changes`
288    pub async fn anomaly(&self, market: impl Into<String>) -> Result<AnomalyResponse> {
289        #[derive(Serialize)]
290        struct Query {
291            market: String,
292            category: &'static str,
293        }
294        self.get(
295            "/v1/quote/changes",
296            Query {
297                market: market.into().to_uppercase(),
298                category: "0",
299            },
300        )
301        .await
302    }
303
304    // ── constituent ───────────────────────────────────────────────
305
306    /// Get constituent stocks for an index.
307    ///
308    /// `symbol` should be an index symbol such as `"HSI.HK"`.
309    ///
310    /// Path: `GET /v1/quote/index-constituents`
311    pub async fn constituent(&self, symbol: impl Into<String>) -> Result<IndexConstituents> {
312        #[derive(Serialize)]
313        struct Query {
314            counter_id: String,
315        }
316        self.get(
317            "/v1/quote/index-constituents",
318            Query {
319                counter_id: index_symbol_to_counter_id(&symbol.into()),
320            },
321        )
322        .await
323    }
324
325    // ── top_movers ────────────────────────────────────────────────
326
327    /// Get top movers (stocks with unusual price movements) across one or more
328    /// markets.
329    ///
330    /// Path: `POST /v1/quote/market/stock-events`
331    ///
332    /// `sort` is the sort order code (0 = ascending, 1 = descending).
333    /// `date` is an optional date filter in `"YYYY-MM-DD"` format.
334    pub async fn top_movers(
335        &self,
336        markets: Vec<String>,
337        sort: u32,
338        date: Option<String>,
339        limit: u32,
340    ) -> Result<TopMoversResponse> {
341        #[derive(Debug, Serialize)]
342        struct Body {
343            limit: u32,
344            sort: u32,
345            markets: Vec<String>,
346            #[serde(skip_serializing_if = "Option::is_none")]
347            date: Option<String>,
348        }
349        let raw: serde_json::Value = self
350            .post(
351                "/v1/quote/market/stock-events",
352                Body {
353                    limit,
354                    sort,
355                    markets,
356                    date,
357                },
358            )
359            .await?;
360
361        let events = raw["events"]
362            .as_array()
363            .cloned()
364            .unwrap_or_default()
365            .into_iter()
366            .map(|ev| {
367                let ts = if let Some(n) = ev["timestamp"].as_i64() {
368                    unix_secs_to_rfc3339(n)
369                } else if let Some(s) = ev["timestamp"].as_str() {
370                    unix_secs_str_to_rfc3339(s)
371                } else {
372                    String::new()
373                };
374                let stock_val = &ev["stock"];
375                let stock = TopMoversStock {
376                    symbol: counter_id_to_symbol(stock_val["counter_id"].as_str().unwrap_or("")),
377                    code: stock_val["code"].as_str().unwrap_or("").to_string(),
378                    name: stock_val["name"].as_str().unwrap_or("").to_string(),
379                    full_name: stock_val["full_name"].as_str().unwrap_or("").to_string(),
380                    change: stock_val["change"].as_str().unwrap_or("").to_string(),
381                    last_done: stock_val["last_done"].as_str().unwrap_or("").to_string(),
382                    market: stock_val["market"].as_str().unwrap_or("").to_string(),
383                    labels: stock_val["labels"]
384                        .as_array()
385                        .map(|arr| {
386                            arr.iter()
387                                .filter_map(|l| l.as_str().map(|s| s.to_string()))
388                                .collect()
389                        })
390                        .unwrap_or_default(),
391                    logo: stock_val["logo"].as_str().unwrap_or("").to_string(),
392                };
393                TopMoversEvent {
394                    timestamp: ts,
395                    alert_reason: ev["alert_reason"].as_str().unwrap_or("").to_string(),
396                    alert_type: ev["alert_type"].as_i64().unwrap_or(0),
397                    stock,
398                    post: ev["post"].clone(),
399                }
400            })
401            .collect();
402        let next_params = raw["next_params"].clone();
403        Ok(TopMoversResponse {
404            events,
405            next_params,
406        })
407    }
408
409    // ── rank_categories ───────────────────────────────────────────
410
411    /// Get all available rank category keys and labels.
412    ///
413    /// Path: `GET /v1/quote/market/rank/categories`
414    pub async fn rank_categories(&self) -> Result<RankCategoriesResponse> {
415        #[derive(Serialize)]
416        struct Empty {}
417        let mut raw: serde_json::Value = self
418            .get("/v1/quote/market/rank/categories", Empty {})
419            .await?;
420        // Strip the "ib_" prefix from all key fields so callers get clean keys
421        // that can be passed back to rank_list without the prefix.
422        if let Some(tags) = raw["first_tags"].as_array_mut() {
423            for tag in tags.iter_mut() {
424                if let Some(k) = tag["key"].as_str() {
425                    let stripped = k.strip_prefix("ib_").unwrap_or(k).to_string();
426                    tag["key"] = serde_json::Value::String(stripped);
427                }
428                if let Some(subs) = tag["second_tags"].as_array_mut() {
429                    for sub in subs.iter_mut() {
430                        if let Some(sk) = sub["key"].as_str() {
431                            let stripped = sk.strip_prefix("ib_").unwrap_or(sk).to_string();
432                            sub["key"] = serde_json::Value::String(stripped);
433                        }
434                    }
435                }
436            }
437        }
438        Ok(RankCategoriesResponse { data: raw })
439    }
440
441    // ── rank_list ─────────────────────────────────────────────────
442
443    /// Get a ranked list of securities for the given category key.
444    ///
445    /// Path: `GET /v1/quote/market/rank/list`
446    pub async fn rank_list(
447        &self,
448        key: impl Into<String>,
449        need_article: bool,
450    ) -> Result<RankListResponse> {
451        #[derive(Serialize)]
452        struct Query {
453            key: String,
454            delay_bmp: &'static str,
455            need_article: &'static str,
456        }
457        let key_str = key.into();
458        // Add "ib_" prefix if the caller passed a clean key (without it).
459        let api_key = if key_str.starts_with("ib_") {
460            key_str
461        } else {
462            format!("ib_{key_str}")
463        };
464        let raw: serde_json::Value = self
465            .get(
466                "/v1/quote/market/rank/list",
467                Query {
468                    key: api_key,
469                    delay_bmp: "false",
470                    need_article: if need_article { "true" } else { "false" },
471                },
472            )
473            .await?;
474        let bmp = raw["bmp"].as_bool().unwrap_or(false);
475        let lists = raw["lists"]
476            .as_array()
477            .cloned()
478            .unwrap_or_default()
479            .into_iter()
480            .map(|item| RankListItem {
481                symbol: counter_id_to_symbol(item["counter_id"].as_str().unwrap_or("")),
482                code: item["code"].as_str().unwrap_or("").to_string(),
483                name: item["name"].as_str().unwrap_or("").to_string(),
484                last_done: item["last_done"].as_str().unwrap_or("").to_string(),
485                chg: item["chg"].as_str().unwrap_or("").to_string(),
486                change: item["change"].as_str().unwrap_or("").to_string(),
487                inflow: item["inflow"].as_str().unwrap_or("").to_string(),
488                market_cap: item["market_cap"].as_str().unwrap_or("").to_string(),
489                industry: item["industry"].as_str().unwrap_or("").to_string(),
490                pre_post_price: item["pre_post_price"].as_str().unwrap_or("").to_string(),
491                pre_post_chg: item["pre_post_chg"].as_str().unwrap_or("").to_string(),
492                amplitude: item["amplitude"].as_str().unwrap_or("").to_string(),
493                five_day_chg: item["five_day_chg"].as_str().unwrap_or("").to_string(),
494                turnover_rate: item["turnover_rate"].as_str().unwrap_or("").to_string(),
495                volume_rate: item["volume_rate"].as_str().unwrap_or("").to_string(),
496                pb_ttm: item["pb_ttm"].as_str().unwrap_or("").to_string(),
497            })
498            .collect();
499        Ok(RankListResponse { bmp, lists })
500    }
501}
longbridge/market/context.rs

longbridge/market/
context.rs
