futu_cache/
indicator.rs

1use std::sync::Arc;
2
3use futu_domain_qot_indicator::local_catalog::{LocalIndicatorCatalog, MergedIndicatorCatalog};
4use futu_domain_qot_indicator::{
5    CatalogPushResult, IndicatorApplyError, IndicatorCatalogSnapshot, IndicatorCatalogState,
6    IndicatorInfo, IndicatorInputValue, IndicatorLanguage, IndicatorParam, IndicatorPushOp,
7    IndicatorRefreshOwner, IndicatorRefreshTicket, RefreshProgress,
8};
9use futu_indicator_engine::python::protocol::WireParameter;
10use futu_indicator_engine::python::{
11    PythonGenerationAuthority, PythonParameterKind, PythonScriptLimits,
12    ServerPythonGenerationLease, ServerPythonProgramMetadata, inspect_python_program_contract,
13    python_formula_identity,
14};
15use parking_lot::Mutex;
16
17#[derive(Debug)]
18pub struct IndicatorMetadataCache {
19    state: Mutex<IndicatorCatalogState>,
20    local_catalog: Result<Arc<LocalIndicatorCatalog>, Arc<str>>,
21}
22
23impl Default for IndicatorMetadataCache {
24    fn default() -> Self {
25        Self {
26            state: Mutex::new(IndicatorCatalogState::new()),
27            local_catalog: LocalIndicatorCatalog::load_embedded()
28                .map_err(|error| Arc::<str>::from(error.to_string())),
29        }
30    }
31}
32
33impl IndicatorMetadataCache {
34    #[must_use]
35    pub fn new() -> Arc<Self> {
36        Arc::new(Self::default())
37    }
38
39    #[must_use]
40    pub fn snapshot(&self) -> Option<Arc<IndicatorCatalogSnapshot>> {
41        self.state.lock().snapshot()
42    }
43
44    pub fn local_catalog(&self) -> Result<Arc<LocalIndicatorCatalog>, Arc<str>> {
45        self.local_catalog.clone()
46    }
47
48    pub fn merged_snapshot(&self) -> Result<MergedIndicatorCatalog, Arc<str>> {
49        self.local_catalog()
50            .map(|local| local.merge_server(self.snapshot()))
51    }
52
53    pub fn begin_refresh(&self, owner: IndicatorRefreshOwner) -> IndicatorRefreshTicket {
54        self.state.lock().begin_refresh(owner)
55    }
56
57    pub fn apply_info_page(
58        &self,
59        ticket: IndicatorRefreshTicket,
60        language: IndicatorLanguage,
61        items: Vec<IndicatorInfo>,
62        next_start_guid: Option<&str>,
63    ) -> Result<RefreshProgress, IndicatorApplyError> {
64        self.state
65            .lock()
66            .apply_info_page(ticket, language, items, next_start_guid)
67    }
68
69    pub fn apply_param_page(
70        &self,
71        ticket: IndicatorRefreshTicket,
72        language: IndicatorLanguage,
73        items: Vec<IndicatorParam>,
74        next_start_guid: Option<&str>,
75    ) -> Result<RefreshProgress, IndicatorApplyError> {
76        self.state
77            .lock()
78            .apply_param_page(ticket, language, items, next_start_guid)
79    }
80
81    pub fn fail_stream(
82        &self,
83        ticket: IndicatorRefreshTicket,
84        language: IndicatorLanguage,
85        info_stream: bool,
86    ) -> bool {
87        self.state.lock().fail_stream(ticket, language, info_stream)
88    }
89
90    pub fn apply_info_push(
91        &self,
92        owner: IndicatorRefreshOwner,
93        op: IndicatorPushOp,
94        item: IndicatorInfo,
95    ) -> Result<CatalogPushResult, IndicatorApplyError> {
96        self.state.lock().apply_info_push(owner, op, item)
97    }
98
99    pub fn apply_param_push(
100        &self,
101        owner: IndicatorRefreshOwner,
102        op: IndicatorPushOp,
103        item: IndicatorParam,
104    ) -> Result<CatalogPushResult, IndicatorApplyError> {
105        self.state.lock().apply_param_push(owner, op, item)
106    }
107}
108
109impl PythonGenerationAuthority for IndicatorMetadataCache {
110    fn resolve_current(&self, guid: &str) -> Option<ServerPythonProgramMetadata> {
111        let snapshot = self.snapshot()?;
112        let record = snapshot.record(IndicatorLanguage::Python, guid)?;
113        let info = record.info.as_ref()?;
114        let script = info.script.as_ref()?.clone();
115        let contract =
116            inspect_python_program_contract(&script, PythonScriptLimits::DEFAULT).ok()?;
117        let parameter_values = match record.param.as_ref() {
118            Some(parameter_metadata) if !parameter_metadata.input_values.is_empty() => {
119                if contract.parameters.len() != parameter_metadata.input_values.len() {
120                    return None;
121                }
122                contract
123                    .parameters
124                    .iter()
125                    .zip(&parameter_metadata.input_values)
126                    .map(|(contract, value)| {
127                        wire_parameter(contract.kind, &contract.var_name, value)
128                    })
129                    .collect::<Option<Vec<_>>>()?
130            }
131            Some(_) | None => Vec::new(),
132        };
133        Some(ServerPythonProgramMetadata {
134            publication_generation: snapshot.publication_generation,
135            mutation_revision: snapshot.mutation_revision,
136            owner_auth_generation: snapshot.owner.auth_generation,
137            owner_connection_generation: snapshot.owner.connection_generation,
138            guid: guid.into(),
139            formula_identity: python_formula_identity(&script),
140            script,
141            parameters: contract.parameters,
142            parameter_values,
143            declared_outputs: contract.declared_outputs,
144        })
145    }
146
147    fn is_current(&self, lease: &ServerPythonGenerationLease) -> bool {
148        self.snapshot().is_some_and(|snapshot| {
149            snapshot.publication_generation == lease.publication_generation()
150                && snapshot.mutation_revision == lease.mutation_revision()
151                && snapshot.owner.auth_generation == lease.owner_auth_generation()
152                && snapshot.owner.connection_generation == lease.owner_connection_generation()
153                && snapshot
154                    .record(IndicatorLanguage::Python, lease.guid())
155                    .and_then(|record| record.info.as_ref())
156                    .and_then(|info| info.script.as_deref())
157                    .is_some_and(|script| {
158                        python_formula_identity(script) == lease.formula_identity()
159                    })
160        })
161    }
162}
163
164fn wire_parameter(
165    kind: PythonParameterKind,
166    name: &str,
167    value: &IndicatorInputValue,
168) -> Option<WireParameter> {
169    Some(match (kind, value) {
170        (PythonParameterKind::Integer, IndicatorInputValue::Int(value)) => {
171            WireParameter::integer(name, *value)
172        }
173        (PythonParameterKind::Float, IndicatorInputValue::Float(value)) => {
174            let value = value
175                .parse::<f64>()
176                .ok()
177                .filter(|value| value.is_finite())?;
178            WireParameter::float(name, value)
179        }
180        (PythonParameterKind::String, IndicatorInputValue::String(value)) => {
181            WireParameter::string(name, value)
182        }
183        (PythonParameterKind::Bool, IndicatorInputValue::Bool(value)) => {
184            WireParameter::boolean(name, *value)
185        }
186        (PythonParameterKind::Color, IndicatorInputValue::Color(value)) => {
187            WireParameter::color(name, (*value as u32) as i64)
188        }
189        (PythonParameterKind::Shape, IndicatorInputValue::Shape(value)) => {
190            WireParameter::shape(name, *value)
191        }
192        (PythonParameterKind::Line, IndicatorInputValue::Line(value)) => {
193            WireParameter::line(name, *value)
194        }
195        _ => return None,
196    })
197}
futu_cache/indicator.rs

futu_cache/
indicator.rs
