use super::{
AvailabilityContext, IndicatorAvailability, IndicatorDefinition, IndicatorDefinitionError,
IndicatorHistoryRequirement, IndicatorInputRequirement, IndicatorInputSource, IndicatorSetting,
IndicatorSettingValue, IndicatorSettingsContext, IndicatorUnsupportedReason,
};
use crate::chart::Basis;
use crate::util::stable_hash;
use exchange::UnixMs;
use indicator_engine::CompiledIndicator;
use rustc_hash::{FxHashMap, FxHashSet};
use serde::{Deserialize, Serialize};
use std::fmt::{Display, Formatter};
use std::sync::Arc;
use std::sync::OnceLock;
static CATALOG: OnceLock<IndicatorCatalog> = OnceLock::new();
static EMPTY_CATALOG: std::sync::LazyLock<IndicatorCatalog> =
std::sync::LazyLock::new(|| IndicatorCatalog::from_registrations(Vec::new()));
pub fn catalog() -> &'static IndicatorCatalog {
CATALOG.get().unwrap_or(&EMPTY_CATALOG)
}
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct IndicatorId(indicator_engine::IndicatorId);
impl IndicatorId {
pub fn new(id: impl Into<String>) -> Result<Self, IndicatorIdError> {
indicator_engine::IndicatorId::new(id)
.map(Self)
.map_err(IndicatorIdError)
}
pub fn parse(id: &str) -> Result<Self, IndicatorIdError> {
Self::new(id)
}
pub fn id(&self) -> &str {
self.0.as_str()
}
pub fn as_str(&self) -> &str {
self.id()
}
pub fn default_instance_slot(&self) -> u64 {
stable_hash(self.id().as_bytes()) & 0x0000_7fff_ffff_ffff
}
pub fn is_registered(&self) -> bool {
catalog().get(self).is_some()
}
pub fn definition(&self) -> &'static IndicatorDefinition {
catalog()
.get(self)
.map(|registration| ®istration.definition)
.unwrap_or_else(missing_definition)
}
pub fn default_config(&self) -> IndicatorConfig {
catalog()
.get(self)
.map(|registration| registration.default_config())
.unwrap_or_default()
}
fn normalized_config(&self, config: IndicatorConfig) -> IndicatorConfig {
catalog()
.get(self)
.map(|registration| registration.normalize_config(config.clone()))
.unwrap_or(config)
}
pub fn settings(
&self,
config: &IndicatorConfig,
context: IndicatorSettingsContext,
) -> Result<Vec<IndicatorSetting>, IndicatorConfigError> {
catalog()
.get(self)
.ok_or_else(|| IndicatorConfigError::MissingIndicator(self.clone()))
.and_then(|registration| registration.settings(config.clone(), context))
}
pub fn with_setting(
&self,
config: IndicatorConfig,
id: &str,
value: IndicatorSettingValue,
) -> Option<IndicatorConfig> {
catalog()
.get(self)
.and_then(|registration| registration.with_setting(config, id, value))
}
pub fn history_requirement(
&self,
config: IndicatorConfig,
basis: Basis,
reference_time: UnixMs,
) -> IndicatorHistoryRequirement {
catalog()
.get(self)
.map(|registration| registration.history_requirement(config, basis, reference_time))
.unwrap_or_default()
}
pub(crate) fn history_requirement_with_warmup(
&self,
config: IndicatorConfig,
basis: Basis,
reference_time: UnixMs,
warmup_bars: Option<u64>,
) -> IndicatorHistoryRequirement {
catalog()
.get(self)
.map(|registration| {
registration.history_requirement_with_warmup(
config,
basis,
reference_time,
warmup_bars,
)
})
.unwrap_or_default()
}
pub fn warmup_bars(&self, config: &IndicatorConfig) -> Option<u64> {
catalog()
.get(self)
.and_then(|registration| registration.warmup_bars(config))
}
pub fn unsupported_reason(
&self,
context: AvailabilityContext,
) -> Option<IndicatorUnsupportedReason> {
let Some(registration) = catalog().get(self) else {
return Some(IndicatorUnsupportedReason::NotInstalled);
};
registration.unsupported_reason(context)
}
pub fn availability(&self, context: AvailabilityContext) -> IndicatorAvailability {
if let Some(reason) = self.unsupported_reason(context) {
return IndicatorAvailability::Unsupported(reason);
}
IndicatorAvailability::Available
}
pub fn declared_input_requirements(&self) -> Vec<IndicatorInputRequirement> {
self.definition().inputs.to_vec()
}
pub fn output_channels(&self) -> &'static [super::OutputChannelSpec] {
catalog()
.get(self)
.map(|registration| registration.definition.outputs.as_slice())
.unwrap_or(&[])
}
pub fn default_view(&self) -> &'static super::IndicatorPanelRecipe {
self.definition().default_view()
}
pub fn requires_trade_data(&self) -> bool {
self.definition().requires(IndicatorInputSource::Trades)
}
pub fn color_specs(&self) -> &'static [super::IndicatorColorSpec] {
catalog()
.get(self)
.map(|registration| registration.definition.style_slots.as_slice())
.unwrap_or(&[])
}
pub fn mark_specs(&self) -> &'static [super::IndicatorMarkSpec] {
catalog()
.get(self)
.map(|registration| registration.definition.mark_specs.as_slice())
.unwrap_or(&[])
}
pub fn segment_specs(&self) -> &'static [super::IndicatorSegmentSpec] {
catalog()
.get(self)
.map(|registration| registration.definition.segment_specs.as_slice())
.unwrap_or(&[])
}
pub fn region_style_slots(&self) -> &'static [String] {
catalog()
.get(self)
.map(|registration| registration.definition.region_style_slots.as_slice())
.unwrap_or(&[])
}
pub fn mark_source_channel(&self, mark: super::MarkKind) -> Option<&'static str> {
self.mark_spec(mark)
.map(|spec| spec.source_channel.as_str())
}
pub fn mark_spec(&self, mark: super::MarkKind) -> Option<&'static super::IndicatorMarkSpec> {
self.mark_specs().iter().find(|spec| spec.mark == mark)
}
pub fn fallback_mark_for(&self, mark: super::MarkKind) -> Option<super::MarkKind> {
self.mark_spec(mark).and_then(|spec| spec.fallback_mark)
}
pub fn overlay_channels(&self) -> &'static [super::IndicatorOverlayChannelSpec] {
catalog()
.get(self)
.map(|registration| registration.definition.overlay_channels.as_slice())
.unwrap_or(&[])
}
pub fn readout_specs(&self) -> &'static [super::IndicatorReadoutSpec] {
match self.default_view() {
super::IndicatorPanelRecipe::Plot { readouts, .. } => readouts,
_ => &[],
}
}
pub fn fill_specs(&self) -> &'static [super::IndicatorFillSpec] {
catalog()
.get(self)
.map(|registration| registration.definition.fill_specs.as_slice())
.unwrap_or(&[])
}
pub fn mark_uses_style_slot(&self, mark: super::MarkKind, slot: &str) -> bool {
self.mark_specs().iter().any(|spec| {
spec.mark == mark
&& (spec.color_slot == slot
|| spec.true_color_slot.as_deref() == Some(slot)
|| spec.false_color_slot.as_deref() == Some(slot)
|| spec.highlight_color_slot.as_deref() == Some(slot))
})
}
}
impl Display for IndicatorId {
fn fmt(&self, formatter: &mut Formatter<'_>) -> std::fmt::Result {
if let Some(registration) = catalog().get(self) {
return formatter.write_str(®istration.definition.display_name);
}
formatter.write_str(self.id())
}
}
impl Serialize for IndicatorId {
fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
where
S: serde::Serializer,
{
serializer.serialize_str(self.id())
}
}
impl<'de> Deserialize<'de> for IndicatorId {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: serde::Deserializer<'de>,
{
let id = String::deserialize(deserializer)?;
Self::parse(&id).map_err(serde::de::Error::custom)
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct IndicatorIdError(String);
impl Display for IndicatorIdError {
fn fmt(&self, formatter: &mut Formatter<'_>) -> std::fmt::Result {
formatter.write_str(&self.0)
}
}
impl std::error::Error for IndicatorIdError {}
impl std::str::FromStr for IndicatorId {
type Err = IndicatorIdError;
fn from_str(id: &str) -> Result<Self, Self::Err> {
Self::parse(id)
}
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(transparent)]
pub struct IndicatorConfig(serde_json::Value);
impl Default for IndicatorConfig {
fn default() -> Self {
Self(serde_json::Value::Object(serde_json::Map::new()))
}
}
impl IndicatorConfig {
pub fn default_for(indicator: &IndicatorId) -> Self {
indicator.default_config()
}
pub fn from_typed<T: Serialize>(config: &T) -> Result<Self, IndicatorConfigError> {
serde_json::to_value(config)
.map(Self)
.map_err(|error| IndicatorConfigError::Serialize(error.to_string()))
}
pub fn from_value(value: serde_json::Value) -> Self {
Self(value)
}
pub fn into_value(self) -> serde_json::Value {
self.0
}
pub fn value(&self) -> &serde_json::Value {
&self.0
}
pub fn is_valid_for(&self, indicator: &IndicatorId) -> bool {
catalog().get(indicator).is_some_and(|registration| {
registration
.validate_config(self)
.inspect_err(|error| {
log::warn!(
"Rejecting indicator {} configuration: {error}",
indicator.id()
);
})
.is_ok()
})
}
pub fn normalized_for(self, indicator: &IndicatorId) -> Self {
indicator.normalized_config(self)
}
pub fn settings_for(
&self,
indicator: &IndicatorId,
context: IndicatorSettingsContext,
) -> Result<Vec<IndicatorSetting>, IndicatorConfigError> {
indicator.settings(self, context)
}
pub fn with_setting_for(
self,
indicator: &IndicatorId,
id: &str,
value: IndicatorSettingValue,
) -> Option<Self> {
indicator.with_setting(self, id, value)
}
pub fn history_requirement_for(
self,
indicator: &IndicatorId,
basis: Basis,
reference_time: UnixMs,
) -> IndicatorHistoryRequirement {
indicator.history_requirement(self, basis, reference_time)
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum IndicatorConfigError {
MissingIndicator(IndicatorId),
Deserialize(String),
Serialize(String),
InvalidDefinition(IndicatorDefinitionError),
}
impl Display for IndicatorConfigError {
fn fmt(&self, formatter: &mut Formatter<'_>) -> std::fmt::Result {
match self {
Self::MissingIndicator(id) => {
write!(formatter, "indicator {} is not installed", id.id())
}
Self::Deserialize(error) => {
write!(formatter, "invalid indicator configuration: {error}")
}
Self::Serialize(error) => write!(
formatter,
"could not serialize indicator configuration: {error}"
),
Self::InvalidDefinition(error) => {
write!(formatter, "invalid indicator definition: {error:?}")
}
}
}
}
impl std::error::Error for IndicatorConfigError {}
#[derive(Clone)]
pub struct IndicatorRegistration {
pub id: IndicatorId,
pub origin: super::language::PackageOrigin,
pub definition: IndicatorDefinition,
pub(super) compiled: Arc<CompiledIndicator>,
}
impl std::fmt::Debug for IndicatorRegistration {
fn fmt(&self, formatter: &mut Formatter<'_>) -> std::fmt::Result {
formatter
.debug_struct("IndicatorRegistration")
.field("id", &self.id)
.field("origin", &self.origin)
.field("definition", &self.definition)
.finish_non_exhaustive()
}
}
impl IndicatorRegistration {
pub(super) fn from_parts(
id: IndicatorId,
origin: super::language::PackageOrigin,
definition: IndicatorDefinition,
compiled: Arc<CompiledIndicator>,
) -> Self {
Self {
id,
origin,
definition,
compiled,
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum CatalogInstallError {
AlreadyInitialized,
Diagnostics(Vec<String>),
}
impl Display for CatalogInstallError {
fn fmt(&self, formatter: &mut Formatter<'_>) -> std::fmt::Result {
match self {
Self::AlreadyInitialized => {
formatter.write_str("indicator catalog is already initialized")
}
Self::Diagnostics(diagnostics) => write!(formatter, "{}", diagnostics.join("\n")),
}
}
}
impl std::error::Error for CatalogInstallError {}
pub(crate) fn install_catalog_with_bundled_diagnostics(
bundled_registrations: Vec<IndicatorRegistration>,
installed_registrations: Vec<IndicatorRegistration>,
loader_diagnostics: Vec<String>,
) -> Result<&'static IndicatorCatalog, CatalogInstallError> {
let (registrations, loader_diagnostics) = merge_bundled_and_installed(
bundled_registrations,
installed_registrations,
loader_diagnostics,
);
let catalog =
IndicatorCatalog::from_registrations_with_diagnostics(registrations, loader_diagnostics);
for diagnostic in &catalog.diagnostics {
log::warn!("indicator catalog: quarantined source: {diagnostic}");
}
CATALOG
.set(catalog)
.map_err(|_| CatalogInstallError::AlreadyInitialized)?;
Ok(CATALOG.get().expect("catalog was installed"))
}
fn merge_bundled_and_installed(
bundled_registrations: Vec<IndicatorRegistration>,
installed_registrations: Vec<IndicatorRegistration>,
mut diagnostics: Vec<String>,
) -> (Vec<IndicatorRegistration>, Vec<String>) {
let mut registrations = bundled_registrations;
let bundled_ids = registrations
.iter()
.map(|registration| registration.id.clone())
.collect::<FxHashSet<_>>();
let mut seen_ids = bundled_ids.clone();
for registration in installed_registrations {
let id = registration.id.clone();
if !seen_ids.insert(id.clone()) {
let reason = if bundled_ids.contains(&id) {
"it is provided by the bundled catalog"
} else {
"an earlier installed source already provides it"
};
diagnostics.push(format!(
"installed indicator '{}' ignored because {reason}",
id.id()
));
log::warn!(
"indicator catalog: ignoring later installed source '{}'",
id.id()
);
continue;
}
registrations.push(registration);
}
(registrations, diagnostics)
}
#[derive(Debug)]
pub struct IndicatorCatalog {
registrations: Vec<IndicatorRegistration>,
by_id: FxHashMap<IndicatorId, usize>,
diagnostics: Vec<String>,
}
impl IndicatorCatalog {
fn from_registrations(registrations: Vec<IndicatorRegistration>) -> Self {
Self::from_registrations_with_diagnostics(registrations, Vec::new())
}
fn from_registrations_with_diagnostics(
mut registrations: Vec<IndicatorRegistration>,
mut diagnostics: Vec<String>,
) -> Self {
registrations.sort_by(|left, right| left.id.id().cmp(right.id.id()));
let mut valid = Vec::with_capacity(registrations.len());
let mut by_id = FxHashMap::default();
for registration in registrations {
if let Err(error) = IndicatorId::parse(registration.id.id()) {
diagnostics.push(format!(
"{}: invalid indicator ID: {error}",
registration.id.id()
));
continue;
}
let id = registration.id.clone();
if by_id.contains_key(&id) {
diagnostics.push(format!("duplicate indicator ID: {}", registration.id.id()));
continue;
}
let default_config = registration.default_config();
if let Err(error) = registration.validate_config(&default_config) {
diagnostics.push(format!(
"{}: invalid default configuration: {error}",
registration.id.id()
));
continue;
}
let index = valid.len();
valid.push(registration);
by_id.insert(id, index);
}
for diagnostic in &diagnostics {
log::error!("indicator catalog: {diagnostic}");
}
Self {
registrations: valid,
by_id,
diagnostics,
}
}
pub fn registrations(&self) -> impl ExactSizeIterator<Item = &IndicatorRegistration> + '_ {
self.registrations.iter()
}
pub fn ids(&self) -> impl ExactSizeIterator<Item = IndicatorId> + '_ {
self.registrations
.iter()
.map(|registration| registration.id.clone())
}
pub fn get(&self, id: &IndicatorId) -> Option<&IndicatorRegistration> {
self.by_id
.get(id)
.and_then(|index| self.registrations.get(*index))
}
pub fn diagnostics(&self) -> &[String] {
&self.diagnostics
}
}
pub fn all_indicators() -> Vec<IndicatorId> {
catalog().ids().collect()
}
pub fn indicator_definition(indicator: &IndicatorId) -> &'static IndicatorDefinition {
indicator.definition()
}
pub fn declared_input_requirements(indicator: &IndicatorId) -> Vec<IndicatorInputRequirement> {
indicator.declared_input_requirements()
}
pub fn display_name(indicator: &IndicatorId) -> &'static str {
&indicator.definition().display_name
}
pub fn panel_recipe(indicator: &IndicatorId) -> &'static super::IndicatorPanelRecipe {
indicator.default_view()
}
pub fn requires_trade_data(indicator: &IndicatorId) -> bool {
indicator.requires_trade_data()
}
pub fn table_columns(indicator: &IndicatorId) -> &'static [super::TableColumnSpec] {
match &indicator.definition().panel_recipe {
super::IndicatorPanelRecipe::Table { columns, .. } => columns,
_ => &[],
}
}
pub fn unsupported_reason(
indicator: &IndicatorId,
context: AvailabilityContext,
) -> Option<IndicatorUnsupportedReason> {
indicator.unsupported_reason(context)
}
pub fn availability(
indicator: &IndicatorId,
context: AvailabilityContext,
) -> IndicatorAvailability {
indicator.availability(context)
}
fn missing_definition() -> &'static IndicatorDefinition {
static MISSING_DEFINITION: OnceLock<IndicatorDefinition> = OnceLock::new();
MISSING_DEFINITION.get_or_init(|| {
IndicatorDefinition::new(
"Unavailable indicator",
Vec::new(),
Vec::new(),
super::IndicatorPanelRecipe::Plot {
panel_title: "Unavailable indicator".to_owned(),
layer_name: "Unavailable indicator".to_owned(),
value_precision: super::PanelValuePrecision::FixedStep(1.0),
value_label_policy: super::PanelValueLabelPolicy {
axis_mode: super::super::composition::PanelValueLabelMode::Compact,
header_mode: super::super::composition::PanelValueLabelMode::Commas,
max_decimals: None,
},
preferred_scale: super::PanelScaleMode::FitVisible,
layers: Vec::new(),
readouts: Vec::new(),
},
)
})
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn ids_validate_publisher_namespaces() {
for id in ["community.ema-cross", "core.custom", "catalog.rsi"] {
let parsed = IndicatorId::parse(id).expect("valid publisher ID");
let serialized = serde_json::to_string(&parsed).unwrap();
assert_eq!(
serde_json::from_str::<IndicatorId>(&serialized).unwrap(),
parsed
);
}
for id in ["ema-cross", "Community.EMA"] {
assert!(IndicatorId::parse(id).is_err());
assert!(serde_json::from_str::<IndicatorId>(&format!("{id:?}")).is_err());
}
}
#[test]
fn duplicate_registrations_are_excluded() {
let scripted = super::super::language::repository_registrations()
.into_iter()
.find(|registration| registration.id.id() == "core.rsi")
.expect("repository RSI registration");
let registrations = vec![scripted.clone(), scripted];
let catalog = IndicatorCatalog::from_registrations(registrations);
assert_eq!(catalog.registrations().len(), 1);
assert!(
catalog
.diagnostics()
.iter()
.any(|diagnostic| diagnostic.contains("duplicate"))
);
}
#[test]
fn bundled_wins_and_installed_sources_keep_first_registration() {
let registrations = super::super::language::repository_registrations();
let core = registrations
.iter()
.find(|registration| registration.id.id() == "core.volume")
.expect("repository Volume registration")
.clone();
let rsi = registrations
.iter()
.find(|registration| registration.id.id() == "core.rsi")
.expect("repository RSI registration");
let mut first_installed = rsi.clone();
first_installed.origin = super::super::language::PackageOrigin::Installed;
first_installed.definition = IndicatorDefinition {
display_name: "First installed registration".to_owned(),
..first_installed.definition
};
let mut later_installed = rsi.clone();
later_installed.origin = super::super::language::PackageOrigin::Installed;
later_installed.definition = IndicatorDefinition {
display_name: "Later installed registration".to_owned(),
..later_installed.definition
};
let (merged, diagnostics) = merge_bundled_and_installed(
vec![core.clone()],
vec![core, first_installed, later_installed],
Vec::new(),
);
let catalog = IndicatorCatalog::from_registrations_with_diagnostics(merged, diagnostics);
assert_eq!(
catalog
.registrations()
.map(|registration| registration.id.id())
.collect::<Vec<_>>(),
vec!["core.rsi", "core.volume"]
);
assert_eq!(
catalog
.get(&IndicatorId::parse("core.volume").unwrap())
.expect("core Volume should be registered")
.definition
.display_name,
"Volume"
);
assert_eq!(
catalog
.get(&IndicatorId::parse("core.rsi").unwrap())
.expect("first installed RSI should be registered")
.definition
.display_name,
"First installed registration"
);
assert!(
catalog
.diagnostics()
.iter()
.any(|diagnostic| diagnostic.contains("provided by the bundled catalog"))
);
assert!(
catalog
.diagnostics()
.iter()
.any(|diagnostic| diagnostic
.contains("earlier installed source already provides it"))
);
}
}