backtest_server/

instrument_catalog.rs

use std::collections::{BTreeMap, BTreeSet};
use std::fs;
use std::sync::Arc;

use chrono::{DateTime, NaiveDateTime, Utc};
use qs_backtest::{
    ReplayInstrumentArtifact, ReplayInstrumentManifest, guarded_instrument_spec,
    resolve_legacy_economics,
};
use qs_instruments::{
    AssetId, AssetKind, AssetSpec, CatalogDocument, Decimal, DecimalGrid, EconomicsModelId,
    EffectiveInterval, InstrumentAlias, InstrumentAssets, InstrumentCatalogSnapshot,
    InstrumentEconomics, InstrumentId, InstrumentResolutionContext, InstrumentResolutionError,
    InstrumentSelector, InstrumentSpec, ListingStatus, ListingVenueId, MarketDataSourceId,
    MarketKind, PriceRules, QuantityRules, QuantityUnit, StoredSeriesBinding,
};
use qs_symbols::{SymbolCurrencyMetadata, SymbolRegistry, SymbolSpec};

use crate::config::{InstrumentsSection, LinearInstrumentConfig};
use crate::error::{BacktestServerError, Result};
use crate::rpc_types::InstrumentExclusionReasonMsg;
use qs_market_loader::SymbolPartitionResolver;

const CATALOG_SCHEMA_VERSION: u32 = 1;
const COMPATIBILITY_LISTING_VENUE: &str = "repository-default";
const REGISTRY_CATALOG_VERSION: &str = "symbol-registry-1";
const SPEC_VALID_FROM: &str = "1970-01-01T00:00:00Z";

#[derive(Clone)]
pub struct InstrumentDomain {
    snapshot: Arc<InstrumentCatalogSnapshot>,
    default_listing_venue: Option<ListingVenueId>,
    data_source: MarketDataSourceId,
    source_symbols: BTreeMap<String, String>,
}

impl InstrumentDomain {
    pub fn load(config: &InstrumentsSection, registry: &SymbolRegistry) -> Result<Self> {
        let configured_listing_venue = config
            .default_listing_venue
            .as_deref()
            .map(ListingVenueId::new)
            .transpose()
            .map_err(|error| BacktestServerError::Config(error.to_string()))?;
        let data_source = MarketDataSourceId::new(&config.market_data_source)
            .map_err(|error| BacktestServerError::Config(error.to_string()))?;
        if config.catalog_path.is_some() && !config.linear_instruments.is_empty() {
            return Err(BacktestServerError::Config(
                "catalog_path and linear_instruments are alternative specification authorities"
                    .into(),
            ));
        }
        let mut source_symbols = BTreeMap::new();
        let mut physical = BTreeSet::new();
        for (symbol, source) in &config.source_symbols {
            let canonical = registry.normalize(symbol).ok_or_else(|| {
                BacktestServerError::Config(format!("unknown source binding symbol '{symbol}'"))
            })?;
            if source.is_empty()
                || source.contains(['/', '\\'])
                || source == "."
                || source == ".."
                || registry
                    .normalize(source)
                    .is_some_and(|target| target != canonical)
                || !physical.insert(source.to_ascii_lowercase())
                || source_symbols
                    .insert(canonical.into(), source.clone())
                    .is_some()
            {
                return Err(BacktestServerError::Config(format!(
                    "invalid or duplicate source binding for '{symbol}'"
                )));
            }
        }
        let (snapshot, default_listing_venue) = match &config.catalog_path {
            Some(path) => {
                let content = fs::read_to_string(path).map_err(|error| {
                    BacktestServerError::Config(format!(
                        "failed to read instrument catalog '{path}': {error}"
                    ))
                })?;
                let document = toml::from_str::<CatalogDocument>(&content).map_err(|error| {
                    BacktestServerError::Config(format!(
                        "failed to parse instrument catalog '{path}': {error}"
                    ))
                })?;
                let snapshot = InstrumentCatalogSnapshot::compile(document).map_err(|error| {
                    BacktestServerError::Config(format!(
                        "invalid instrument catalog '{path}': {error}"
                    ))
                })?;
                (snapshot, configured_listing_venue)
            }
            None => {
                let listing_venue = configured_listing_venue.unwrap_or_else(|| {
                    ListingVenueId::new(COMPATIBILITY_LISTING_VENUE)
                        .expect("valid built-in compatibility listing venue")
                });
                (
                    compatibility_snapshot(registry, &listing_venue, &config.linear_instruments)?,
                    Some(listing_venue),
                )
            }
        };
        Ok(Self {
            snapshot: Arc::new(snapshot),
            default_listing_venue,
            data_source,
            source_symbols,
        })
    }

    pub fn compatibility(registry: &SymbolRegistry) -> Result<Self> {
        Self::load(&InstrumentsSection::default(), registry)
    }

    pub fn snapshot(&self) -> &InstrumentCatalogSnapshot {
        &self.snapshot
    }

    pub fn data_source(&self) -> &MarketDataSourceId {
        &self.data_source
    }

    pub fn symbol_resolver<'a>(
        &'a self,
        registry: &'a SymbolRegistry,
    ) -> SymbolPartitionResolver<'a> {
        SymbolPartitionResolver::with_source_symbols(registry, &self.source_symbols)
    }

    pub fn resolve_manifest(
        &self,
        symbols: &[String],
        at: NaiveDateTime,
        through: Option<NaiveDateTime>,
    ) -> Result<ReplayInstrumentManifest> {
        let allowed_instruments = self
            .snapshot
            .instrument_ids()
            .cloned()
            .collect::<BTreeSet<_>>();
        let context = InstrumentResolutionContext {
            allowed_instruments,
            default_listing_venue: self.default_listing_venue.clone(),
            default_market_kind: None,
        };
        let at = at.and_utc();
        let through = through.map(|value| value.and_utc());
        let mut instruments = BTreeMap::new();
        for symbol in symbols {
            let selector = InstrumentSelector::Alias {
                alias: InstrumentAlias::new(symbol).map_err(|error| {
                    BacktestServerError::InvalidRequest(format!(
                        "invalid instrument selector for '{symbol}': {error}"
                    ))
                })?,
                listing_venue: None,
                market_kind: None,
            };
            let resolved = self
                .snapshot
                .resolve(&selector, &context, at)
                .map_err(|error| resolution_error(symbol, error, at))?;
            validate_replay_spec(symbol, &resolved.spec)?;
            if let Some(through) = through {
                let end = self
                    .snapshot
                    .resolve(&selector, &context, through)
                    .map_err(|error| resolution_error(symbol, error, through))?;
                if end.reference != resolved.reference {
                    return Err(BacktestServerError::InstrumentUnavailable {
                        symbol: symbol.clone(),
                        reason: InstrumentExclusionReasonMsg::UnsupportedEconomics,
                        details:
                            "instrument changes specification during the requested replay range"
                                .into(),
                    });
                }
            }
            instruments.insert(
                symbol.clone(),
                ReplayInstrumentArtifact {
                    resolved: resolved.reference,
                    spec: resolved.spec.as_ref().clone(),
                },
            );
        }
        Ok(ReplayInstrumentManifest {
            instruments,
            stored_series: Vec::new(),
        })
    }

    pub fn attach_stored_series(
        &self,
        manifest: &mut ReplayInstrumentManifest,
        coordinates: impl IntoIterator<Item = (String, String, String)>,
    ) -> Result<()> {
        let mut bindings = Vec::new();
        for (symbol, source_partition, source_symbol) in coordinates {
            let artifact = manifest.instruments.get(&symbol).ok_or_else(|| {
                BacktestServerError::InvalidRequest(format!(
                    "stored series for '{symbol}' has no resolved instrument"
                ))
            })?;
            bindings.push(StoredSeriesBinding {
                data_source: self.data_source.clone(),
                source_partition,
                source_symbol,
                instrument: artifact.resolved.clone(),
                effective: artifact.spec.effective,
            });
        }
        bindings.sort_by(|left, right| {
            left.source_partition
                .cmp(&right.source_partition)
                .then(left.source_symbol.cmp(&right.source_symbol))
                .then(left.instrument.instrument.cmp(&right.instrument.instrument))
        });
        bindings.dedup();
        manifest.stored_series = bindings;
        Ok(())
    }
}

fn resolution_error(
    symbol: &str,
    error: InstrumentResolutionError,
    at: DateTime<Utc>,
) -> BacktestServerError {
    let details = format!("cannot resolve instrument '{symbol}': {error}");
    match error {
        InstrumentResolutionError::Inactive { .. } => BacktestServerError::InactiveInstrument {
            symbol: symbol.into(),
            at,
            details,
        },
        InstrumentResolutionError::Ambiguous { .. } => BacktestServerError::InstrumentUnavailable {
            symbol: symbol.into(),
            reason: InstrumentExclusionReasonMsg::AmbiguousMapping,
            details,
        },
        _ => BacktestServerError::InstrumentUnavailable {
            symbol: symbol.into(),
            reason: InstrumentExclusionReasonMsg::UnknownInstrument,
            details,
        },
    }
}

fn compatibility_snapshot(
    registry: &SymbolRegistry,
    listing_venue: &ListingVenueId,
    linear_instruments: &[LinearInstrumentConfig],
) -> Result<InstrumentCatalogSnapshot> {
    let effective = EffectiveInterval::new(
        SPEC_VALID_FROM
            .parse::<DateTime<Utc>>()
            .expect("valid built-in instrument epoch"),
        None,
    )
    .expect("valid built-in instrument interval");
    let mut assets = BTreeMap::<AssetId, AssetSpec>::new();
    let mut instruments = Vec::new();
    for (symbol, currencies) in registry.entries() {
        let Ok(economics) = resolve_legacy_economics(symbol) else {
            continue;
        };
        register_assets(&mut assets, symbol, currencies)?;
        let instrument = InstrumentId::new(
            listing_venue.clone(),
            market_kind(symbol)?,
            symbol.canonical.parse().map_err(|error| {
                BacktestServerError::Config(format!(
                    "invalid listing ID for '{}': {error}",
                    symbol.canonical
                ))
            })?,
        );
        instruments.push(
            guarded_instrument_spec(symbol, currencies, economics, instrument, effective)
                .map_err(|error| BacktestServerError::Config(error.to_string()))?,
        );
    }
    let mut configured = BTreeSet::new();
    for rules in linear_instruments {
        let canonical = registry.normalize(&rules.symbol).ok_or_else(|| {
            BacktestServerError::Config(format!("unknown linear instrument '{}'", rules.symbol))
        })?;
        if !configured.insert(canonical.to_owned()) {
            return Err(BacktestServerError::Config(format!(
                "duplicate linear instrument '{canonical}'"
            )));
        }
        let symbol = registry
            .spec(canonical)
            .expect("normalized registered symbol");
        let currencies = registry.currency_metadata(canonical).ok_or_else(|| {
            BacktestServerError::Config(format!("missing currencies for '{canonical}'"))
        })?;
        register_assets(&mut assets, symbol, currencies)?;
        let settlement = AssetId::new(&currencies.pnl_currency)
            .map_err(|error| BacktestServerError::Config(error.to_string()))?;
        let spec = InstrumentSpec {
            revision: "1.0.0"
                .parse()
                .map_err(|error: qs_instruments::IdentifierError| {
                    BacktestServerError::Config(error.to_string())
                })?,
            instrument: InstrumentId::new(
                listing_venue.clone(),
                MarketKind::new(MarketKind::LINEAR_EXPOSURE)
                    .map_err(|error| BacktestServerError::Config(error.to_string()))?,
                canonical
                    .parse()
                    .map_err(|error: qs_instruments::IdentifierError| {
                        BacktestServerError::Config(error.to_string())
                    })?,
            ),
            effective,
            status: ListingStatus::Trading,
            assets: InstrumentAssets {
                base: currencies
                    .base_currency
                    .as_deref()
                    .map(AssetId::new)
                    .transpose()
                    .map_err(|error| BacktestServerError::Config(error.to_string()))?,
                quote: currencies
                    .quote_currency
                    .as_deref()
                    .map(AssetId::new)
                    .transpose()
                    .map_err(|error| BacktestServerError::Config(error.to_string()))?,
                settlement: settlement.clone(),
                fee_assets: BTreeSet::new(),
            },
            price: PriceRules {
                grid: DecimalGrid::new(Decimal::ZERO, rules.price_step),
                display_scale: rules.display_scale,
            },
            quantity: QuantityRules {
                grid: DecimalGrid::new(Decimal::ZERO, rules.quantity_step),
                minimum: rules.minimum,
                maximum: Some(rules.maximum),
                storage_scale: rules
                    .quantity_step
                    .get()
                    .scale()
                    .max(rules.minimum.get().scale())
                    .max(rules.maximum.get().scale()),
            },
            notional: None,
            economics: InstrumentEconomics {
                pnl_model: EconomicsModelId::new(EconomicsModelId::CFD_QUOTE_LINEAR_V1)
                    .map_err(|error| BacktestServerError::Config(error.to_string()))?,
                quantity_unit: QuantityUnit::StandardLot,
                contract_multiplier: rules.contract_multiplier,
                settlement_asset: settlement,
                fee_model: None,
                funding_model: None,
                margin_model: None,
            },
            aliases: BTreeSet::from([InstrumentAlias::new(canonical)
                .map_err(|error| BacktestServerError::Config(error.to_string()))?]),
        };
        spec.validate()
            .map_err(|error| BacktestServerError::Config(error.to_string()))?;
        instruments.retain(|existing| {
            !existing
                .aliases
                .contains(&InstrumentAlias::new(canonical).expect("validated alias"))
        });
        instruments.push(spec);
    }
    InstrumentCatalogSnapshot::compile(CatalogDocument {
        schema_version: CATALOG_SCHEMA_VERSION,
        version: REGISTRY_CATALOG_VERSION.into(),
        assets: assets.into_values().collect(),
        instruments,
    })
    .map_err(|error| BacktestServerError::Config(error.to_string()))
}

fn register_assets(
    assets: &mut BTreeMap<AssetId, AssetSpec>,
    symbol: &SymbolSpec,
    currencies: &SymbolCurrencyMetadata,
) -> Result<()> {
    let values = [
        currencies.base_currency.as_deref(),
        currencies.quote_currency.as_deref(),
        Some(currencies.pnl_currency.as_str()),
    ];
    for value in values.into_iter().flatten() {
        let asset =
            AssetId::new(value).map_err(|error| BacktestServerError::Config(error.to_string()))?;
        let kind = match symbol.category.as_str() {
            "crypto" if currencies.base_currency.as_deref() == Some(value) => AssetKind::Crypto,
            "metal" | "commodity" if currencies.base_currency.as_deref() == Some(value) => {
                AssetKind::Commodity
            }
            _ => AssetKind::Fiat,
        };
        assets.entry(asset.clone()).or_insert(AssetSpec {
            asset,
            kind,
            display_code: value.to_ascii_uppercase(),
            storage_scale: None,
        });
    }
    Ok(())
}

fn market_kind(symbol: &SymbolSpec) -> Result<MarketKind> {
    let kind = match symbol.category.as_str() {
        "forex" => MarketKind::FX_CFD,
        "metal" => MarketKind::METAL_CFD,
        "commodity" => MarketKind::COMMODITY_CFD,
        "index" => MarketKind::INDEX_CFD,
        category => {
            return Err(BacktestServerError::Config(format!(
                "unsupported registry category '{category}' for {}",
                symbol.canonical
            )));
        }
    };
    MarketKind::new(kind).map_err(|error| BacktestServerError::Config(error.to_string()))
}

fn validate_replay_spec(symbol: &str, spec: &qs_instruments::InstrumentSpec) -> Result<()> {
    if spec.status != ListingStatus::Trading {
        return Err(BacktestServerError::InstrumentUnavailable {
            symbol: symbol.into(),
            reason: InstrumentExclusionReasonMsg::UnsupportedEconomics,
            details: "instrument is not in trading status".into(),
        });
    }
    if spec.economics.quantity_unit != QuantityUnit::StandardLot {
        return Err(BacktestServerError::InstrumentUnavailable {
            symbol: symbol.into(),
            reason: InstrumentExclusionReasonMsg::UnsupportedEconomics,
            details: "unsupported quantity unit".into(),
        });
    }
    let model = spec.economics.pnl_model.as_str();
    if model != EconomicsModelId::FX_QUOTE_LINEAR_V1
        && model != EconomicsModelId::CFD_QUOTE_LINEAR_V1
    {
        return Err(BacktestServerError::InstrumentUnavailable {
            symbol: symbol.into(),
            reason: InstrumentExclusionReasonMsg::UnsupportedEconomics,
            details: format!("unsupported P&L model: {model}"),
        });
    }
    Ok(())
}

#[cfg(test)]
mod tests {
    use std::collections::BTreeSet;
    use std::time::{SystemTime, UNIX_EPOCH};

    use qs_instruments::{
        Decimal, DecimalGrid, InstrumentAssets, InstrumentEconomics, InstrumentSpec, PriceRules,
        QuantityRules,
    };

    use super::*;

    fn registry() -> SymbolRegistry {
        SymbolRegistry::from_toml(
            r#"
[[symbol]]
canonical = "eurusd"
aliases = ["eur/usd"]
pip_position = 4
digits = 5
category = "forex"
base_currency = "EUR"
quote_currency = "USD"
pnl_currency = "USD"
lot_base_units = 100000
lot_step_units = 1000

[[symbol]]
canonical = "btcusd"
aliases = ["btc/usd"]
pip_position = 1
digits = 2
category = "crypto"
base_currency = "BTC"
quote_currency = "USD"
pnl_currency = "USD"
lot_base_units = 100000000
lot_step_units = 100000
"#,
        )
        .unwrap()
    }

    #[test]
    fn compatibility_catalog_uses_owned_listing_namespace_not_data_platform() {
        let domain = InstrumentDomain::compatibility(&registry()).unwrap();
        let manifest = domain
            .resolve_manifest(
                &["eurusd".into()],
                "2026-01-01T00:00:00".parse().unwrap(),
                None,
            )
            .unwrap();
        let instrument = &manifest.instruments["eurusd"].resolved.instrument;
        assert_eq!(instrument.listing_venue.as_str(), "repository-default");
        assert_ne!(instrument.listing_venue.as_str(), "ctrader");
        assert_eq!(instrument.market_kind.as_str(), MarketKind::FX_CFD);
    }

    #[test]
    fn unsupported_registry_rows_do_not_enter_the_compatibility_catalog() {
        let domain = InstrumentDomain::compatibility(&registry()).unwrap();
        let error = domain
            .resolve_manifest(
                &["btcusd".into()],
                "2026-01-01T00:00:00".parse().unwrap(),
                None,
            )
            .unwrap_err();
        assert!(error.to_string().contains("cannot resolve instrument"));
    }

    #[test]
    fn explicit_catalog_loader_preserves_broker_listing_and_rejects_unknown_fields() {
        let usd: AssetId = "USD".parse().unwrap();
        let document = CatalogDocument {
            schema_version: 1,
            version: "broker-catalog-1".into(),
            assets: vec![
                AssetSpec {
                    asset: "EUR".parse().unwrap(),
                    kind: AssetKind::Fiat,
                    display_code: "EUR".into(),
                    storage_scale: Some(2),
                },
                AssetSpec {
                    asset: usd.clone(),
                    kind: AssetKind::Fiat,
                    display_code: "USD".into(),
                    storage_scale: Some(2),
                },
            ],
            instruments: vec![InstrumentSpec {
                revision: "1.0.0".parse().unwrap(),
                instrument: InstrumentId::new(
                    "ic-markets".parse().unwrap(),
                    MarketKind::new(MarketKind::FX_CFD).unwrap(),
                    "EURUSD".parse().unwrap(),
                ),
                effective: EffectiveInterval::new("2026-01-01T00:00:00Z".parse().unwrap(), None)
                    .unwrap(),
                status: ListingStatus::Trading,
                assets: InstrumentAssets {
                    base: Some("EUR".parse().unwrap()),
                    quote: Some(usd.clone()),
                    settlement: usd.clone(),
                    fee_assets: BTreeSet::new(),
                },
                price: PriceRules {
                    grid: DecimalGrid::new(Decimal::ZERO, "0.00001".parse().unwrap()),
                    display_scale: 5,
                },
                quantity: QuantityRules {
                    grid: DecimalGrid::new(Decimal::ZERO, "0.01".parse().unwrap()),
                    minimum: "0.01".parse().unwrap(),
                    maximum: Some("100".parse().unwrap()),
                    storage_scale: 2,
                },
                notional: None,
                economics: InstrumentEconomics {
                    pnl_model: EconomicsModelId::new(EconomicsModelId::FX_QUOTE_LINEAR_V1).unwrap(),
                    quantity_unit: QuantityUnit::StandardLot,
                    contract_multiplier: "100000".parse().unwrap(),
                    settlement_asset: usd,
                    fee_model: None,
                    funding_model: None,
                    margin_model: None,
                },
                aliases: BTreeSet::from(["EURUSD".parse().unwrap()]),
            }],
        };
        let unique = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .unwrap()
            .as_nanos();
        let path = std::env::temp_dir().join(format!(
            "qs_instrument_catalog_{}_{}.toml",
            std::process::id(),
            unique
        ));
        let content = toml::to_string(&document).unwrap();
        fs::write(&path, &content).unwrap();
        let config = InstrumentsSection {
            catalog_path: Some(path.to_string_lossy().into_owned()),
            default_listing_venue: None,
            market_data_source: "test-parquet".into(),
            ..InstrumentsSection::default()
        };

        let domain = InstrumentDomain::load(&config, &registry()).unwrap();
        let manifest = domain
            .resolve_manifest(
                &["eurusd".into()],
                "2026-02-01T00:00:00".parse().unwrap(),
                None,
            )
            .unwrap();
        let instrument = &manifest.instruments["eurusd"].resolved.instrument;
        assert_eq!(instrument.listing_venue.as_str(), "ic-markets");
        assert_ne!(instrument.listing_venue.as_str(), "ctrader");

        fs::write(&path, format!("unexpected = true\n{content}")).unwrap();
        let error = match InstrumentDomain::load(&config, &registry()) {
            Ok(_) => panic!("catalog with an unknown field must be rejected"),
            Err(error) => error,
        };
        assert!(error.to_string().contains("unknown field"));
        fs::remove_file(path).unwrap();
    }
}