longbridge/dca/
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::{Config, Result, dca::types::*, utils::counter::symbol_to_counter_id};
8
9struct InnerDCAContext {
10    http_cli: HttpClient,
11    log_subscriber: Arc<dyn Subscriber + Send + Sync>,
12}
13
14impl Drop for InnerDCAContext {
15    fn drop(&mut self) {
16        dispatcher::with_default(&self.log_subscriber.clone().into(), || {
17            tracing::info!("dca context dropped");
18        });
19    }
20}
21
22/// Dollar-cost averaging (DCA) plan management context.
23#[derive(Clone)]
24pub struct DCAContext(Arc<InnerDCAContext>);
25
26impl DCAContext {
27    /// Create a [`DCAContext`]
28    pub fn new(config: Arc<Config>) -> Self {
29        let log_subscriber = config.create_log_subscriber("dca");
30        dispatcher::with_default(&log_subscriber.clone().into(), || {
31            tracing::info!(language = ?config.language, "creating dca context");
32        });
33        let ctx = Self(Arc::new(InnerDCAContext {
34            http_cli: config.create_http_client(),
35            log_subscriber,
36        }));
37        dispatcher::with_default(&ctx.0.log_subscriber.clone().into(), || {
38            tracing::info!("dca context created");
39        });
40        ctx
41    }
42
43    /// Returns the log subscriber
44    #[inline]
45    pub fn log_subscriber(&self) -> Arc<dyn Subscriber + Send + Sync> {
46        self.0.log_subscriber.clone()
47    }
48
49    async fn get<R, Q>(&self, path: &'static str, query: Q) -> Result<R>
50    where
51        R: DeserializeOwned + Send + Sync + 'static,
52        Q: Serialize + Send + Sync,
53    {
54        Ok(self
55            .0
56            .http_cli
57            .request(Method::GET, path)
58            // Recurring investment (DCA) is served only by the AP data center.
59            .dc_restrict(DcRegion::Ap)
60            .query_params(query)
61            .response::<Json<R>>()
62            .send()
63            .with_subscriber(self.0.log_subscriber.clone())
64            .await?
65            .0)
66    }
67
68    async fn post<R, B>(&self, path: &'static str, body: B) -> Result<R>
69    where
70        R: DeserializeOwned + Send + Sync + 'static,
71        B: std::fmt::Debug + Serialize + Send + Sync + 'static,
72    {
73        Ok(self
74            .0
75            .http_cli
76            .request(Method::POST, path)
77            // Recurring investment (DCA) is served only by the AP data center.
78            .dc_restrict(DcRegion::Ap)
79            .body(Json(body))
80            .response::<Json<R>>()
81            .send()
82            .with_subscriber(self.0.log_subscriber.clone())
83            .await?
84            .0)
85    }
86
87    /// List DCA plans.
88    ///
89    /// Path: `GET /v1/dailycoins/query`
90    pub async fn list(&self, status: Option<DCAStatus>, symbol: Option<String>) -> Result<DcaList> {
91        #[derive(Serialize)]
92        struct Query {
93            page: i32,
94            limit: i32,
95            #[serde(skip_serializing_if = "Option::is_none")]
96            status: Option<DCAStatus>,
97            #[serde(skip_serializing_if = "Option::is_none")]
98            counter_id: Option<String>,
99        }
100        self.get(
101            "/v1/dailycoins/query",
102            Query {
103                page: 1,
104                limit: 100,
105                status,
106                counter_id: symbol.map(|s| symbol_to_counter_id(&s)),
107            },
108        )
109        .await
110    }
111
112    /// Create a new DCA plan.
113    ///
114    /// Path: `POST /v1/dailycoins/create`
115    pub async fn create(
116        &self,
117        symbol: impl Into<String>,
118        amount: impl Into<String>,
119        frequency: DCAFrequency,
120        day_of_week: Option<String>,
121        day_of_month: Option<u32>,
122        allow_margin: bool,
123    ) -> Result<DcaCreateResult> {
124        let cid = symbol_to_counter_id(&symbol.into());
125        let mut body = serde_json::json!({
126            "counter_id": cid,
127            "per_invest_amount": amount.into(),
128            "invest_frequency": frequency,
129            "allow_margin_finance": if allow_margin { 1 } else { 0 }
130        });
131        if let Some(dow) = day_of_week {
132            body["invest_day_of_week"] = serde_json::Value::String(dow);
133        }
134        if let Some(dom) = day_of_month {
135            body["invest_day_of_month"] = serde_json::Value::String(dom.to_string());
136        }
137        self.post("/v1/dailycoins/create", body).await
138    }
139
140    /// Update a DCA plan.
141    ///
142    /// Path: `POST /v1/dailycoins/update`
143    pub async fn update(
144        &self,
145        plan_id: impl Into<String>,
146        amount: Option<String>,
147        frequency: Option<DCAFrequency>,
148        day_of_week: Option<String>,
149        day_of_month: Option<u32>,
150        allow_margin: Option<bool>,
151    ) -> Result<DcaCreateResult> {
152        let mut body = serde_json::json!({ "plan_id": plan_id.into() });
153        if let Some(a) = amount {
154            body["per_invest_amount"] = serde_json::Value::String(a);
155        }
156        if let Some(f) = frequency {
157            body["invest_frequency"] = serde_json::to_value(f).unwrap_or_default();
158        }
159        if let Some(dow) = day_of_week {
160            body["invest_day_of_week"] = serde_json::Value::String(dow);
161        }
162        if let Some(dom) = day_of_month {
163            body["invest_day_of_month"] = serde_json::Value::String(dom.to_string());
164        }
165        if let Some(m) = allow_margin {
166            body["allow_margin_finance"] =
167                serde_json::Value::Number((if m { 1 } else { 0 }).into());
168        }
169        self.post::<DcaCreateResult, _>("/v1/dailycoins/update", body)
170            .await
171    }
172
173    /// Pause a DCA plan.
174    pub async fn pause(&self, plan_id: impl Into<String>) -> Result<()> {
175        self.post::<serde_json::Value, _>(
176            "/v1/dailycoins/toggle",
177            serde_json::json!({ "plan_id": plan_id.into(), "status": "Suspended" }),
178        )
179        .await?;
180        Ok(())
181    }
182
183    /// Resume a suspended DCA plan.
184    pub async fn resume(&self, plan_id: impl Into<String>) -> Result<()> {
185        self.post::<serde_json::Value, _>(
186            "/v1/dailycoins/toggle",
187            serde_json::json!({ "plan_id": plan_id.into(), "status": "Active" }),
188        )
189        .await?;
190        Ok(())
191    }
192
193    /// Stop (permanently finish) a DCA plan.
194    pub async fn stop(&self, plan_id: impl Into<String>) -> Result<()> {
195        self.post::<serde_json::Value, _>(
196            "/v1/dailycoins/toggle",
197            serde_json::json!({ "plan_id": plan_id.into(), "status": "Finished" }),
198        )
199        .await?;
200        Ok(())
201    }
202
203    /// Get execution history for a DCA plan.
204    ///
205    /// Path: `GET /v1/dailycoins/query-records`
206    pub async fn history(
207        &self,
208        plan_id: impl Into<String>,
209        page: i32,
210        limit: i32,
211    ) -> Result<DcaHistoryResponse> {
212        #[derive(Serialize)]
213        struct Query {
214            plan_id: String,
215            page: i32,
216            limit: i32,
217        }
218        self.get(
219            "/v1/dailycoins/query-records",
220            Query {
221                plan_id: plan_id.into(),
222                page,
223                limit,
224            },
225        )
226        .await
227    }
228
229    /// Get DCA statistics.
230    ///
231    /// Path: `GET /v1/dailycoins/statistic`
232    pub async fn stats(&self, symbol: Option<String>) -> Result<DcaStats> {
233        #[derive(Serialize)]
234        struct Query {
235            #[serde(skip_serializing_if = "Option::is_none")]
236            counter_id: Option<String>,
237        }
238        self.get(
239            "/v1/dailycoins/statistic",
240            Query {
241                counter_id: symbol.map(|s| symbol_to_counter_id(&s)),
242            },
243        )
244        .await
245    }
246
247    /// Check DCA support for a list of securities.
248    ///
249    /// Path: `POST /v1/dailycoins/batch-check-support`
250    pub async fn check_support(&self, symbols: Vec<String>) -> Result<DcaSupportList> {
251        let counter_ids: Vec<String> = symbols.iter().map(|s| symbol_to_counter_id(s)).collect();
252        self.post(
253            "/v1/dailycoins/batch-check-support",
254            serde_json::json!({ "counter_ids": counter_ids }),
255        )
256        .await
257    }
258
259    /// Calculate the next projected trade date for a DCA plan with the given
260    /// schedule parameters.
261    ///
262    /// Path: `POST /v1/dailycoins/calc-trd-date`
263    pub async fn calc_date(
264        &self,
265        symbol: impl Into<String>,
266        frequency: DCAFrequency,
267        day_of_week: Option<String>,
268        day_of_month: Option<u32>,
269    ) -> Result<DcaCalcDateResult> {
270        let mut body = serde_json::json!({
271            "counter_id": symbol_to_counter_id(&symbol.into()),
272            "invest_frequency": frequency,
273        });
274        if let Some(dow) = day_of_week {
275            body["invest_day_of_week"] = serde_json::Value::String(dow);
276        }
277        if let Some(dom) = day_of_month {
278            body["invest_day_of_month"] = serde_json::Value::String(dom.to_string());
279        }
280        self.post("/v1/dailycoins/calc-trd-date", body).await
281    }
282
283    /// Update the advance reminder hours for DCA execution notifications.
284    ///
285    /// `hours` must be one of `"1"`, `"6"`, or `"12"`.
286    ///
287    /// Path: `POST /v1/dailycoins/update-alter-hours`
288    pub async fn set_reminder(&self, hours: impl Into<String>) -> Result<()> {
289        #[derive(serde::Deserialize)]
290        struct Empty {}
291        self.post::<Empty, _>(
292            "/v1/dailycoins/update-alter-hours",
293            serde_json::json!({ "alter_hours": hours.into() }),
294        )
295        .await?;
296        Ok(())
297    }
298}
longbridge/dca/context.rs

longbridge/dca/
context.rs
