futu_gateway_core/
stock_db.rs

1// 股票列表 SQLite 持久化
2//
3// 对齐 C++ SecListDBHelper：将股票列表存储在本地 SQLite 数据库中，
4// 重启时从本地加载，避免全量同步。
5
6use rusqlite::{Connection, Result as SqlResult, Row, params, types::Type};
7use std::path::{Path, PathBuf};
8use std::sync::{Mutex, MutexGuard};
9
10mod schema;
11#[cfg(test)]
12pub(crate) use schema::{CURRENT_SCHEMA_VERSION, write_user_version};
13pub(crate) use schema::{migrate_schema, read_user_version};
14
15/// 数据库文件路径: ~/.futu-opend-rs/stock_data.db
16fn db_path() -> PathBuf {
17    let home = std::env::var("HOME")
18        .or_else(|_| std::env::var("USERPROFILE"))
19        .unwrap_or_else(|_| ".".to_string());
20    let mut path = PathBuf::from(home);
21    path.push(".futu-opend-rs");
22    path.push("stock_data.db");
23    path
24}
25
26/// 存储在 SQLite 中的股票信息
27#[derive(Debug, Clone)]
28pub struct DbStockItem {
29    pub stock_id: u64,
30    /// C++ `security.sequence`; required for CMD6746 seq_check_sum_v2 parity.
31    pub sequence: u64,
32    pub code: String,
33    pub name_sc: String,
34    /// stock_list_sync `CSStockItem.tc_name` field 5.
35    pub name_tc: String,
36    /// stock_list_sync `CSStockItem.eng_name` field 9.
37    pub name_en: String,
38    /// stock_list_sync `CSStockItem.ja_name` field 65.
39    pub name_jp: String,
40    pub market_code: u32,
41    pub instrument_type: u32,
42    /// stock_list_sync `CSStockItem.spread_table_code` field 13.
43    ///
44    /// Required by `GetSecuritySnapshot.priceSpread` to mirror C++
45    /// `INNBiz_Qot_Spread::AdjustPrice`.
46    pub spread_table_code: u32,
47    /// stock_list_sync `CSStockItem.sub_instrument_type_v2` field 69.
48    ///
49    /// Required by v10.6 `GetCompanyProfile` to mirror C++
50    /// `APIServer_Qot_CompanyProfile.cpp:55-63` Trust-vs-ETF branch.
51    pub sub_instrument_type_v2: u32,
52    pub lot_size: u32,
53    pub delisting: bool,
54    pub warrnt_stock_owner: u64,
55    /// C++ internal `NN_QotWarrantType`, stock-list `warrnt_type`.
56    /// Values 6/7 are DLC Long/Short and are filtered from GetReference(WARRANT).
57    pub warrant_type: u32,
58    pub no_search: bool,
59    pub listing_date: u32,
60    /// v1.4.106 codex F5: 期货主连合约关联 stock_id (proto field 41 origin_id).
61    /// schema version 2 加. 老 row 默认 0.
62    pub future_origin_id: u64,
63    /// v1.4.106 codex F5: 期货主力合约 stock_id (proto field 40 zhuli_id).
64    /// schema version 2 加. 老 row 默认 0.
65    pub zhuli_id: u64,
66    /// Crypto 货币对左侧货币 (stock_list_sync `cc_origin`, proto field 60).
67    pub cc_origin: String,
68    /// Crypto 货币对右侧货币 (stock_list_sync `cc_destination`, proto field 61).
69    pub cc_destination: String,
70    /// v1.4.110 final E.5 P2#6: 交易所 (proto field 57). schema v4 加.
71    pub exchange: String,
72    /// v1.4.110 final E.5 P2#6: 上市交易所 (proto field 74). schema v4 加.
73    pub listed_exchange: String,
74}
75
76/// 股票列表数据库 (线程安全，Mutex 包装)
77pub struct StockDb {
78    conn: Mutex<Connection>,
79}
80
81#[derive(Debug, Clone, Copy, PartialEq, Eq)]
82pub struct StockListChecksumV2 {
83    pub id_low: u64,
84    pub id_high: u64,
85    pub seq_low: u64,
86    pub seq_high: u64,
87    pub stock_count: u64,
88}
89
90// SAFETY (v1.4.110 final E.3): `rusqlite::Connection` is `!Send + !Sync` by
91// default because SQLite C handles have implicit thread-affinity in
92// single-threaded compile mode (`SQLITE_THREADSAFE=0`). However, we compile
93// rusqlite with `features = ["bundled"]` (see
94// `crates/futu-gateway-core/Cargo.toml:46`), which links to **bundled SQLite
95// compiled with `SQLITE_THREADSAFE=1` (Serialized mode)**: SQLite's own
96// internal mutex serializes all access to the handle, making it thread-safe
97// to share across threads.
98//
99// We additionally wrap the `Connection` in `std::sync::Mutex<Connection>`
100// which adds Rust-level mutual exclusion. The combination provides:
101//   1. **Send (transfer ownership across threads)**: safe because the
102//      underlying SQLite handle (Serialized mode) does not bind to a specific
103//      thread.
104//   2. **Sync (concurrent shared reference)**: safe because all access is
105//      mediated by the outer `Mutex<Connection>` — only one thread holds the
106//      handle at any time.
107//
108// **Invariants this code relies on**:
109//   (a) rusqlite features list MUST include `"bundled"` (verified by grep on
110//       Cargo.toml — never change without re-evaluating this `unsafe impl`).
111//   (b) `Connection` is ONLY accessed through `self.conn.lock()` (verified by
112//       grep `self\.conn` in this file — every access is `.lock()` first).
113//
114// **Future**: if rusqlite ever exposes a `connection-threadsafe` feature gate
115// or trait that statically asserts Serialized mode, migrate to that and
116// remove these `unsafe impl`s.
117unsafe impl Send for StockDb {}
118unsafe impl Sync for StockDb {}
119
120impl StockDb {
121    #[cfg(test)]
122    pub(crate) fn open_in_memory_for_test() -> SqlResult<Self> {
123        let conn = Connection::open_in_memory()?;
124        conn.execute_batch(
125            "CREATE TABLE IF NOT EXISTS config (
126                key TEXT PRIMARY KEY,
127                value TEXT NOT NULL
128             );
129             CREATE TABLE IF NOT EXISTS securities (
130                stock_id INTEGER PRIMARY KEY,
131                sequence INTEGER NOT NULL DEFAULT 0,
132                code TEXT NOT NULL,
133                name_sc TEXT NOT NULL DEFAULT '',
134                name_tc TEXT NOT NULL DEFAULT '',
135                name_en TEXT NOT NULL DEFAULT '',
136                name_jp TEXT NOT NULL DEFAULT '',
137                market_code INTEGER NOT NULL DEFAULT 0,
138                instrument_type INTEGER NOT NULL DEFAULT 0,
139                spread_table_code INTEGER NOT NULL DEFAULT 0,
140                sub_instrument_type_v2 INTEGER NOT NULL DEFAULT 0,
141                lot_size INTEGER NOT NULL DEFAULT 0,
142                delisting INTEGER NOT NULL DEFAULT 0,
143                warrnt_stock_owner INTEGER NOT NULL DEFAULT 0,
144                warrant_type INTEGER NOT NULL DEFAULT 0,
145                no_search INTEGER NOT NULL DEFAULT 0,
146                listing_date INTEGER NOT NULL DEFAULT 0,
147                future_origin_id INTEGER NOT NULL DEFAULT 0,
148                zhuli_id INTEGER NOT NULL DEFAULT 0,
149                cc_origin TEXT NOT NULL DEFAULT '',
150                cc_destination TEXT NOT NULL DEFAULT '',
151                exchange TEXT NOT NULL DEFAULT '',
152                listed_exchange TEXT NOT NULL DEFAULT ''
153             );",
154        )?;
155        migrate_schema(&conn)?;
156        Ok(Self {
157            conn: Mutex::new(conn),
158        })
159    }
160
161    fn lock_conn(&self) -> SqlResult<MutexGuard<'_, Connection>> {
162        self.conn.lock().map_err(|_| rusqlite::Error::InvalidQuery)
163    }
164
165    fn ensure_parent_dir(parent: &std::path::Path) {
166        if let Err(err) = std::fs::create_dir_all(parent) {
167            tracing::warn!(
168                path = %parent.display(),
169                error = %err,
170                "stock db directory creation failed; sqlite open may fail"
171            );
172        }
173    }
174
175    fn apply_legacy_column(conn: &Connection, column: &'static str, sql: &'static str) {
176        match conn.execute_batch(sql) {
177            Ok(()) => {}
178            Err(err) if is_duplicate_column_error(&err) => {
179                tracing::debug!(column, "legacy stock db column already exists");
180            }
181            Err(err) => {
182                tracing::warn!(
183                    column,
184                    error = %err,
185                    "legacy stock db schema migration failed"
186                );
187            }
188        }
189    }
190
191    /// 打开（或创建）数据库
192    pub fn open() -> SqlResult<Self> {
193        Self::open_at(db_path())
194    }
195
196    /// 打开（或创建）指定路径的数据库。
197    ///
198    /// 生产路径由 [`Self::open`] 统一选择；测试和诊断工具可用此入口避免污染
199    /// 用户真实 `~/.futu-opend-rs/stock_data.db`。
200    pub fn open_at(path: impl AsRef<Path>) -> SqlResult<Self> {
201        let path = path.as_ref();
202        if let Some(parent) = path.parent() {
203            Self::ensure_parent_dir(parent);
204        }
205        let conn = Connection::open(path)?;
206        conn.execute_batch(
207            "PRAGMA journal_mode=WAL;
208             PRAGMA synchronous=NORMAL;
209             PRAGMA cache_size=-8000;",
210        )?;
211        conn.execute_batch(
212            "CREATE TABLE IF NOT EXISTS config (
213                key TEXT PRIMARY KEY,
214                value TEXT NOT NULL
215             );
216             CREATE TABLE IF NOT EXISTS securities (
217                stock_id INTEGER PRIMARY KEY,
218                sequence INTEGER NOT NULL DEFAULT 0,
219                code TEXT NOT NULL,
220                name_sc TEXT NOT NULL DEFAULT '',
221                name_tc TEXT NOT NULL DEFAULT '',
222                name_en TEXT NOT NULL DEFAULT '',
223                name_jp TEXT NOT NULL DEFAULT '',
224                market_code INTEGER NOT NULL DEFAULT 0,
225                instrument_type INTEGER NOT NULL DEFAULT 0,
226                spread_table_code INTEGER NOT NULL DEFAULT 0,
227                sub_instrument_type_v2 INTEGER NOT NULL DEFAULT 0,
228                lot_size INTEGER NOT NULL DEFAULT 0,
229                delisting INTEGER NOT NULL DEFAULT 0,
230                warrnt_stock_owner INTEGER NOT NULL DEFAULT 0,
231                warrant_type INTEGER NOT NULL DEFAULT 0,
232                no_search INTEGER NOT NULL DEFAULT 0,
233                listing_date INTEGER NOT NULL DEFAULT 0,
234                future_origin_id INTEGER NOT NULL DEFAULT 0,
235                zhuli_id INTEGER NOT NULL DEFAULT 0,
236                cc_origin TEXT NOT NULL DEFAULT '',
237                cc_destination TEXT NOT NULL DEFAULT '',
238                exchange TEXT NOT NULL DEFAULT '',
239                listed_exchange TEXT NOT NULL DEFAULT ''
240             );",
241        )?;
242        // 旧版 schema 可能没有 listing_date 列，尝试 ALTER TABLE 补上
243        Self::apply_legacy_column(
244            &conn,
245            "listing_date",
246            "ALTER TABLE securities ADD COLUMN listing_date INTEGER NOT NULL DEFAULT 0;",
247        );
248        // v1.4.113 BUG-008: C++ stock-list cache keeps SC/TC/EN names and
249        // GetStaticInfo selects EN only for AppLanguage_EN.
250        Self::apply_legacy_column(
251            &conn,
252            "name_tc",
253            "ALTER TABLE securities ADD COLUMN name_tc TEXT NOT NULL DEFAULT '';",
254        );
255        Self::apply_legacy_column(
256            &conn,
257            "name_en",
258            "ALTER TABLE securities ADD COLUMN name_en TEXT NOT NULL DEFAULT '';",
259        );
260        // Latest C++ SecListDBHelper VALUE_VERSION=13: stock_list_sync field 65
261        // `ja_name` is persisted as SQLite `name_jp` and used by
262        // GetSecDisplayNameByAppLanguage when NN_AppLanguage_JA.
263        Self::apply_legacy_column(
264            &conn,
265            "name_jp",
266            "ALTER TABLE securities ADD COLUMN name_jp TEXT NOT NULL DEFAULT '';",
267        );
268        // v1.4.106 codex F5: 老版本 schema 可能没有 future_origin_id / zhuli_id
269        // 列, 加 ALTER TABLE 补上 (idempotent — 已存在则 duplicate-column no-op).
270        Self::apply_legacy_column(
271            &conn,
272            "future_origin_id",
273            "ALTER TABLE securities ADD COLUMN future_origin_id INTEGER NOT NULL DEFAULT 0;",
274        );
275        Self::apply_legacy_column(
276            &conn,
277            "zhuli_id",
278            "ALTER TABLE securities ADD COLUMN zhuli_id INTEGER NOT NULL DEFAULT 0;",
279        );
280        Self::apply_legacy_column(
281            &conn,
282            "cc_origin",
283            "ALTER TABLE securities ADD COLUMN cc_origin TEXT NOT NULL DEFAULT '';",
284        );
285        Self::apply_legacy_column(
286            &conn,
287            "cc_destination",
288            "ALTER TABLE securities ADD COLUMN cc_destination TEXT NOT NULL DEFAULT '';",
289        );
290        // v1.4.110 final E.5 P2#6: schema v4 加 exchange / listed_exchange
291        // (proto field 57/74 数据驱动). 老 row 默认空串, 下次 stock_list_sync 重写.
292        Self::apply_legacy_column(
293            &conn,
294            "exchange",
295            "ALTER TABLE securities ADD COLUMN exchange TEXT NOT NULL DEFAULT '';",
296        );
297        Self::apply_legacy_column(
298            &conn,
299            "listed_exchange",
300            "ALTER TABLE securities ADD COLUMN listed_exchange TEXT NOT NULL DEFAULT '';",
301        );
302        Self::apply_legacy_column(
303            &conn,
304            "sub_instrument_type_v2",
305            "ALTER TABLE securities ADD COLUMN sub_instrument_type_v2 INTEGER NOT NULL DEFAULT 0;",
306        );
307        Self::apply_legacy_column(
308            &conn,
309            "spread_table_code",
310            "ALTER TABLE securities ADD COLUMN spread_table_code INTEGER NOT NULL DEFAULT 0;",
311        );
312        Self::apply_legacy_column(
313            &conn,
314            "sequence",
315            "ALTER TABLE securities ADD COLUMN sequence INTEGER NOT NULL DEFAULT 0;",
316        );
317        Self::apply_legacy_column(
318            &conn,
319            "warrant_type",
320            "ALTER TABLE securities ADD COLUMN warrant_type INTEGER NOT NULL DEFAULT 0;",
321        );
322
323        // v1.4.79 B1: SQLite `PRAGMA user_version` 做 schema 版本化。
324        // 首次启动 / 无版本 → user_version=0，migrate_to_current 把我们带到 1。
325        // 未来加字段 bump CURRENT_SCHEMA_VERSION + 加对应 migration step。
326        migrate_schema(&conn)?;
327
328        Ok(Self {
329            conn: Mutex::new(conn),
330        })
331    }
332
333    /// 获取 stock_list_version
334    pub fn get_version(&self) -> u64 {
335        let Ok(conn) = self.lock_conn() else {
336            tracing::warn!(
337                "stock db stock_list_version read failed because connection mutex is poisoned; \
338                 falling back to full stock-list sync"
339            );
340            return 0;
341        };
342        let raw = match conn.query_row(
343            "SELECT value FROM config WHERE key = 'stock_list_version'",
344            [],
345            |row| row.get::<_, String>(0),
346        ) {
347            Ok(raw) => raw,
348            Err(rusqlite::Error::QueryReturnedNoRows) => return 0,
349            Err(err) => {
350                tracing::warn!(
351                    error = %err,
352                    "stock db stock_list_version read failed; falling back to full stock-list sync"
353                );
354                return 0;
355            }
356        };
357        match raw.parse::<u64>() {
358            Ok(version) => version,
359            Err(err) => {
360                tracing::warn!(
361                    value = %raw,
362                    error = %err,
363                    "stock db stock_list_version parse failed; falling back to full stock-list sync"
364                );
365                0
366            }
367        }
368    }
369
370    /// 保存 stock_list_version
371    pub fn set_version(&self, version: u64) -> SqlResult<()> {
372        let conn = self.lock_conn()?;
373        conn.execute(
374            "INSERT OR REPLACE INTO config (key, value) VALUES ('stock_list_version', ?1)",
375            params![version.to_string()],
376        )?;
377        Ok(())
378    }
379
380    /// 加载所有股票记录
381    pub fn load_all(&self) -> SqlResult<Vec<DbStockItem>> {
382        let conn = self.lock_conn()?;
383        let mut stmt = conn.prepare(
384            "SELECT stock_id, code, name_sc, name_tc, name_en, name_jp, market_code, instrument_type,
385                    spread_table_code, sub_instrument_type_v2, lot_size,
386                    delisting, warrnt_stock_owner, warrant_type, no_search, listing_date,
387                    future_origin_id, zhuli_id, cc_origin, cc_destination,
388                    exchange, listed_exchange, sequence
389             FROM securities",
390        )?;
391        let rows = stmt.query_map([], db_stock_item_from_row)?;
392
393        let mut items = Vec::new();
394        let mut skipped_rows = 0usize;
395        for (row_index, row) in rows.enumerate() {
396            match row {
397                Ok(item) => items.push(item),
398                Err(err) => {
399                    skipped_rows += 1;
400                    if skipped_rows <= 5 {
401                        tracing::warn!(
402                            row_index,
403                            error = %err,
404                            "stock db load_all skipped malformed cached row"
405                        );
406                    }
407                }
408            }
409        }
410        if skipped_rows > 0 {
411            tracing::warn!(
412                skipped_rows,
413                loaded_rows = items.len(),
414                "stock db load_all completed with skipped malformed rows"
415            );
416        }
417        Ok(items)
418    }
419
420    /// Load all cached rows with the given public code.
421    ///
422    /// Code is not globally unique across markets, so callers must still apply
423    /// market projection/filtering before using the result.
424    pub fn load_by_code(&self, code: &str) -> SqlResult<Vec<DbStockItem>> {
425        let conn = self.lock_conn()?;
426        let mut stmt = conn.prepare_cached(
427            "SELECT stock_id, code, name_sc, name_tc, name_en, name_jp, market_code, instrument_type,
428                    spread_table_code, sub_instrument_type_v2, lot_size,
429                    delisting, warrnt_stock_owner, warrant_type, no_search, listing_date,
430                    future_origin_id, zhuli_id, cc_origin, cc_destination,
431                    exchange, listed_exchange, sequence
432             FROM securities
433             WHERE code = ?1",
434        )?;
435        let rows = stmt.query_map(params![code], db_stock_item_from_row)?;
436        let mut items = Vec::new();
437        let mut skipped_rows = 0usize;
438        for (row_index, row) in rows.enumerate() {
439            match row {
440                Ok(item) => items.push(item),
441                Err(err) => {
442                    skipped_rows += 1;
443                    if skipped_rows <= 5 {
444                        tracing::warn!(
445                            row_index,
446                            code,
447                            error = %err,
448                            "stock db load_by_code skipped malformed cached row"
449                        );
450                    }
451                }
452            }
453        }
454        if skipped_rows > 0 {
455            tracing::warn!(
456                skipped_rows,
457                loaded_rows = items.len(),
458                code,
459                "stock db load_by_code completed with skipped malformed rows"
460            );
461        }
462        Ok(items)
463    }
464
465    /// C++ `SQLite3SecListDBHelper::SearchWarrantByOwnerID` equivalent:
466    /// `SELECT * FROM security WHERE warrnt_owner = ? AND delete_flag=0`.
467    ///
468    /// Rust stores stock-list delete rows by physically removing them, so the
469    /// remaining table already represents `delete_flag=0`. Do not add a public
470    /// market filter here: C++ `GetReference(Warrant)` returns cross-market
471    /// associated warrants such as SG rows for HK.00700 when they can be
472    /// projected to a public `QotMarket`.
473    pub fn load_warrants_by_owner(&self, owner_stock_id: u64) -> SqlResult<Vec<DbStockItem>> {
474        let conn = self.lock_conn()?;
475        let mut stmt = conn.prepare_cached(
476            "SELECT stock_id, code, name_sc, name_tc, name_en, name_jp, market_code, instrument_type,
477                    spread_table_code, sub_instrument_type_v2, lot_size,
478                    delisting, warrnt_stock_owner, warrant_type, no_search, listing_date,
479                    future_origin_id, zhuli_id, cc_origin, cc_destination,
480                    exchange, listed_exchange, sequence
481             FROM securities
482             WHERE warrnt_stock_owner = ?1",
483        )?;
484        let rows = stmt.query_map(params![owner_stock_id as i64], db_stock_item_from_row)?;
485        let mut items = Vec::new();
486        let mut skipped_rows = 0usize;
487        for (row_index, row) in rows.enumerate() {
488            match row {
489                Ok(item) => items.push(item),
490                Err(err) => {
491                    skipped_rows += 1;
492                    if skipped_rows <= 5 {
493                        tracing::warn!(
494                            row_index,
495                            owner_stock_id,
496                            error = %err,
497                            "stock db load_warrants_by_owner skipped malformed cached row"
498                        );
499                    }
500                }
501            }
502        }
503        if skipped_rows > 0 {
504            tracing::warn!(
505                skipped_rows,
506                loaded_rows = items.len(),
507                owner_stock_id,
508                "stock db load_warrants_by_owner completed with skipped malformed rows"
509            );
510        }
511        Ok(items)
512    }
513
514    /// 开始批量写入事务
515    pub fn begin_batch(&self) -> SqlResult<()> {
516        let conn = self.lock_conn()?;
517        conn.execute_batch("BEGIN TRANSACTION")?;
518        Ok(())
519    }
520
521    /// 提交事务
522    pub fn commit_batch(&self) -> SqlResult<()> {
523        let conn = self.lock_conn()?;
524        conn.execute_batch("COMMIT")?;
525        Ok(())
526    }
527
528    /// 尝试回滚当前批量事务。
529    ///
530    /// `stock_list_sync` 在网络请求前打开批量事务；若 CMD6746 中途失败，
531    /// 需要显式回滚，避免下一轮同步碰到悬挂事务。SQLite 在没有活动事务时
532    /// 会返回错误，caller 负责按上下文记录。
533    pub fn rollback_batch_if_open(&self) -> SqlResult<()> {
534        let conn = self.lock_conn()?;
535        conn.execute_batch("ROLLBACK")?;
536        Ok(())
537    }
538
539    /// 插入或更新一条股票记录
540    pub fn upsert(&self, item: &DbStockItem) -> SqlResult<()> {
541        let conn = self.lock_conn()?;
542        let mut stmt = conn.prepare_cached(
543            "INSERT OR REPLACE INTO securities
544             (stock_id, sequence, code, name_sc, name_tc, name_en, name_jp, market_code, instrument_type,
545              spread_table_code, lot_size,
546              delisting, warrnt_stock_owner, warrant_type, no_search, listing_date,
547              future_origin_id, zhuli_id, cc_origin, cc_destination,
548              exchange, listed_exchange, sub_instrument_type_v2)
549             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, ?17, ?18, ?19, ?20, ?21, ?22, ?23)",
550        )?;
551        stmt.execute(params![
552            item.stock_id as i64,
553            item.sequence as i64,
554            &item.code,
555            &item.name_sc,
556            &item.name_tc,
557            &item.name_en,
558            &item.name_jp,
559            item.market_code as i32,
560            item.instrument_type as i32,
561            item.spread_table_code as i32,
562            item.lot_size as i32,
563            item.delisting as i32,
564            item.warrnt_stock_owner as i64,
565            item.warrant_type as i32,
566            item.no_search as i32,
567            item.listing_date as i32,
568            // v1.4.106 codex F5: schema v2 加
569            item.future_origin_id as i64,
570            item.zhuli_id as i64,
571            &item.cc_origin,
572            &item.cc_destination,
573            // v1.4.110 final E.5 P2#6: schema v4 加 exchange + listed_exchange
574            &item.exchange,
575            &item.listed_exchange,
576            // v10.6.6608 QOT CompanyProfile: C++ enSubTypeV2 branch source.
577            item.sub_instrument_type_v2 as i32,
578        ])?;
579        Ok(())
580    }
581
582    /// 删除一条股票记录
583    pub fn delete(&self, stock_id: u64) -> SqlResult<()> {
584        let conn = self.lock_conn()?;
585        let mut stmt = conn.prepare_cached("DELETE FROM securities WHERE stock_id = ?1")?;
586        stmt.execute(params![stock_id as i64])?;
587        Ok(())
588    }
589
590    /// 删除全部股票记录。
591    ///
592    /// 仅用于 CMD6746 `stock_list_version=0` 全量重建事务。普通 delta sync
593    /// 必须继续依赖服务端 delete_flag，避免把增量页误当全量替换。
594    pub fn clear_all_securities(&self) -> SqlResult<()> {
595        let conn = self.lock_conn()?;
596        conn.execute("DELETE FROM securities", [])?;
597        Ok(())
598    }
599
600    /// 读取当前持久化的全部 stock_id。
601    ///
602    /// full refresh 成功提交后，调用方用这个列表从内存 static cache 删除
603    /// 服务端全量未再返回的旧 row。
604    pub fn load_stock_ids(&self) -> SqlResult<Vec<u64>> {
605        let conn = self.lock_conn()?;
606        let mut stmt = conn.prepare("SELECT stock_id FROM securities")?;
607        let rows = stmt.query_map([], |row| row_u64(row, 0, "stock_id"))?;
608        let mut ids = Vec::new();
609        for row in rows {
610            ids.push(row?);
611        }
612        Ok(ids)
613    }
614
615    /// 获取记录总数
616    pub fn count(&self) -> u64 {
617        let Ok(conn) = self.lock_conn() else {
618            tracing::warn!(
619                "stock db count read failed because connection mutex is poisoned; returning 0"
620            );
621            return 0;
622        };
623        match conn.query_row("SELECT COUNT(*) FROM securities", [], |row| {
624            row.get::<_, i64>(0)
625        }) {
626            Ok(count) if count >= 0 => count as u64,
627            Ok(count) => {
628                tracing::warn!(
629                    count,
630                    "stock db count read returned negative value; returning 0"
631                );
632                0
633            }
634            Err(err) => {
635                tracing::warn!(error = %err, "stock db count read failed; returning 0");
636                0
637            }
638        }
639    }
640
641    /// C++ `SQLite3SecListDBHelper::CalcSecInfoCheckSumV2` split SUM:
642    /// SUM(stock_id low32/high32) and SUM(sequence low32/high32).
643    pub fn calc_checksum_v2(&self) -> SqlResult<StockListChecksumV2> {
644        let conn = self.lock_conn()?;
645        let mut stmt = conn.prepare("SELECT stock_id, sequence FROM securities")?;
646        let rows = stmt.query_map([], |row| {
647            Ok((row_u64(row, 0, "stock_id")?, row_u64(row, 1, "sequence")?))
648        })?;
649        let mut checksum = StockListChecksumV2 {
650            id_low: 0,
651            id_high: 0,
652            seq_low: 0,
653            seq_high: 0,
654            stock_count: 0,
655        };
656        for row in rows {
657            let (stock_id, sequence) = row?;
658            checksum.id_low = checksum.id_low.wrapping_add(stock_id & 0xFFFF_FFFF);
659            checksum.id_high = checksum.id_high.wrapping_add(stock_id >> 32);
660            checksum.seq_low = checksum.seq_low.wrapping_add(sequence & 0xFFFF_FFFF);
661            checksum.seq_high = checksum.seq_high.wrapping_add(sequence >> 32);
662            checksum.stock_count = checksum.stock_count.wrapping_add(1);
663        }
664        Ok(checksum)
665    }
666
667    /// v1.4.79 B1: 读取 SQLite schema version（调试 / daemon startup log）。
668    pub fn schema_version(&self) -> u32 {
669        let Ok(conn) = self.lock_conn() else {
670            tracing::warn!(
671                "stock db schema version read failed because connection mutex is poisoned; \
672                 returning 0"
673            );
674            return 0;
675        };
676        match read_user_version(&conn) {
677            Ok(version) => version,
678            Err(err) => {
679                tracing::warn!(error = %err, "stock db schema version read failed; returning 0");
680                0
681            }
682        }
683    }
684}
685
686fn db_stock_item_from_row(row: &Row<'_>) -> SqlResult<DbStockItem> {
687    Ok(DbStockItem {
688        stock_id: row_u64(row, 0, "stock_id")?,
689        sequence: row_u64(row, 22, "sequence")?,
690        code: row.get(1)?,
691        name_sc: row.get(2)?,
692        name_tc: row.get(3)?,
693        name_en: row.get(4)?,
694        name_jp: row.get(5)?,
695        market_code: row_u32(row, 6, "market_code")?,
696        instrument_type: row_u32(row, 7, "instrument_type")?,
697        spread_table_code: row_u32(row, 8, "spread_table_code")?,
698        sub_instrument_type_v2: row_u32(row, 9, "sub_instrument_type_v2")?,
699        lot_size: row_u32(row, 10, "lot_size")?,
700        delisting: row.get::<_, i32>(11)? != 0,
701        warrnt_stock_owner: row_u64(row, 12, "warrnt_stock_owner")?,
702        warrant_type: row_u32(row, 13, "warrant_type")?,
703        no_search: row.get::<_, i32>(14)? != 0,
704        listing_date: row_u32(row, 15, "listing_date")?,
705        // v1.4.106 codex F5: schema v2 加, 老 row 默认 0.
706        future_origin_id: row_u64(row, 16, "future_origin_id")?,
707        zhuli_id: row_u64(row, 17, "zhuli_id")?,
708        cc_origin: row.get(18)?,
709        cc_destination: row.get(19)?,
710        // v1.4.110 final E.5 P2#6: schema v4 加, 老 row 默认 ''.
711        exchange: row.get(20)?,
712        listed_exchange: row.get(21)?,
713    })
714}
715
716fn row_u64(row: &Row<'_>, idx: usize, field: &'static str) -> SqlResult<u64> {
717    let value = row.get::<_, i64>(idx)?;
718    u64::try_from(value).map_err(|err| {
719        rusqlite::Error::FromSqlConversionFailure(
720            idx,
721            Type::Integer,
722            Box::new(std::io::Error::new(
723                std::io::ErrorKind::InvalidData,
724                format!("stock db field {field} must be non-negative: {err}"),
725            )),
726        )
727    })
728}
729
730fn row_u32(row: &Row<'_>, idx: usize, field: &'static str) -> SqlResult<u32> {
731    let value = row.get::<_, i32>(idx)?;
732    u32::try_from(value).map_err(|err| {
733        rusqlite::Error::FromSqlConversionFailure(
734            idx,
735            Type::Integer,
736            Box::new(std::io::Error::new(
737                std::io::ErrorKind::InvalidData,
738                format!("stock db field {field} must be non-negative: {err}"),
739            )),
740        )
741    })
742}
743
744fn is_duplicate_column_error(err: &rusqlite::Error) -> bool {
745    err.to_string().contains("duplicate column name")
746}
747
748#[cfg(test)]
749mod tests;
futu_gateway_core/stock_db.rs

futu_gateway_core/
stock_db.rs
