use std::cell::RefCell;
use std::collections::{BTreeMap, BTreeSet};
use chrono::NaiveDateTime;
use qs_core::types::{Effect, PositionStatus};
use qs_core::{ManagementProfile, RuleConfigDef, StoplossMode, TradeEngine};
use qs_strategy::{
CommandFact, CommandFeedback, CommandTerminalStatus, ConfiguredActionKind, ConfiguredCommand,
ConfiguredStrategy, ConfiguredStrategyRequirements, DecisionKind, MAX_GENERATED_ID_BYTES,
MAX_ID_BYTES, MAX_NAMED_VALUES, MAX_OUTPUT_COMMANDS, MAX_OUTPUT_NOTES, MAX_TEXT_BYTES,
NamedValue, NoteKind, OutputScalar, SourceId, StrategyInput, TradeSlotFacts, TradeSlotState,
Value, ValueType,
};
use crate::future_executor::FutureExecutor;
use crate::ledger::ActionDispositionStatus;
use crate::portfolio::CampaignExcursion;
use crate::profile::PreparedEntryProfiles;
use super::{
BarSeriesSpec, ClosedBar, HistoricalObservationView, HistoricalSeriesView, JournalKind,
SeriesId, StrategyDecisionDraft, StrategyDecisionKind, StrategyDescriptor, StrategyDomainError,
StrategyFeedbackEvent, StrategyJournalDraft, StrategyJournalError, StrategyObservation,
StrategyRequirements, StrategyResearchLimits, StrategyRetentionLimits,
};
const MAX_EXACT_F64_INTEGER: u64 = 1_u64 << 53;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum HistoricalVolumeProjection {
TickCountExact,
OptionalTickCount,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ConfiguredSourceBinding {
source: SourceId,
series: BarSeriesSpec,
}
impl ConfiguredSourceBinding {
pub fn new(source: SourceId, series: BarSeriesSpec) -> Self {
Self { source, series }
}
pub fn source(&self) -> &SourceId {
&self.source
}
pub fn series(&self) -> &BarSeriesSpec {
&self.series
}
pub fn series_id(&self) -> &SeriesId {
self.series.requirement().id()
}
}
#[derive(Clone, Copy)]
pub struct NamedInputProjectionContext<'a> {
pub observed_through: NaiveDateTime,
pub closed_bars: &'a [ClosedBar],
pub observations: &'a [StrategyObservation],
pub series: &'a dyn HistoricalSeriesView,
pub observation_history: &'a dyn HistoricalObservationView,
}
#[derive(Debug, Clone, PartialEq)]
pub struct ProjectedNamedInput {
pub value: Value,
pub updated: bool,
}
pub trait HistoricalNamedInputProjector: Send {
fn output_type(&self) -> ValueType;
fn project(
&self,
context: NamedInputProjectionContext<'_>,
) -> Result<ProjectedNamedInput, NamedInputProjectionError>;
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SourceBarFactKind {
Ordinal,
OpenTime,
CloseTime,
AvailableAt,
GapBefore,
}
#[derive(Debug, Clone, Copy, Default)]
struct SourceBarFactState {
last_open: Option<NaiveDateTime>,
last_close: Option<NaiveDateTime>,
ordinal: u64,
}
pub struct SourceBarFactProjector {
series_id: SeriesId,
kind: SourceBarFactKind,
state: RefCell<SourceBarFactState>,
}
impl SourceBarFactProjector {
pub fn new(series_id: SeriesId, kind: SourceBarFactKind) -> Self {
Self {
series_id,
kind,
state: RefCell::new(SourceBarFactState::default()),
}
}
}
impl HistoricalNamedInputProjector for SourceBarFactProjector {
fn output_type(&self) -> ValueType {
ValueType::optional(match self.kind {
SourceBarFactKind::Ordinal => qs_strategy::ScalarType::Integer,
SourceBarFactKind::OpenTime
| SourceBarFactKind::CloseTime
| SourceBarFactKind::AvailableAt => qs_strategy::ScalarType::Timestamp,
SourceBarFactKind::GapBefore => qs_strategy::ScalarType::Bool,
})
}
fn project(
&self,
context: NamedInputProjectionContext<'_>,
) -> Result<ProjectedNamedInput, NamedInputProjectionError> {
let Some(bar) = context
.closed_bars
.iter()
.find(|bar| bar.series_id() == &self.series_id)
else {
return Ok(ProjectedNamedInput {
value: Value::Missing(self.output_type().scalar),
updated: false,
});
};
let mut state = self.state.borrow_mut();
let is_new = state.last_open != Some(bar.open_time());
let previous_close = state.last_close;
if is_new {
state.ordinal = state
.ordinal
.checked_add(1)
.ok_or_else(|| NamedInputProjectionError::new("source ordinal overflowed"))?;
state.last_open = Some(bar.open_time());
state.last_close = Some(bar.close_time());
}
let value = match self.kind {
SourceBarFactKind::Ordinal => Value::Integer(
i64::try_from(state.ordinal)
.map_err(|_| NamedInputProjectionError::new("source ordinal exceeds i64"))?,
),
SourceBarFactKind::OpenTime => Value::Timestamp(bar.open_time()),
SourceBarFactKind::CloseTime => Value::Timestamp(bar.close_time()),
SourceBarFactKind::AvailableAt => Value::Timestamp(context.observed_through),
SourceBarFactKind::GapBefore => {
Value::Bool(previous_close.is_some_and(|close| close != bar.open_time()))
}
};
Ok(ProjectedNamedInput {
value,
updated: is_new,
})
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ConfirmedSwingFactKind {
Price,
AnchorOpenTime,
AnchorCloseTime,
ConfirmedAt,
}
pub struct ConfirmedSwingFactProjector {
series_id: SeriesId,
swing_kind: super::SwingKind,
fact: ConfirmedSwingFactKind,
retained: RefCell<Option<(u64, Value)>>,
}
impl ConfirmedSwingFactProjector {
pub fn new(
series_id: SeriesId,
swing_kind: super::SwingKind,
fact: ConfirmedSwingFactKind,
) -> Self {
Self {
series_id,
swing_kind,
fact,
retained: RefCell::new(None),
}
}
}
impl HistoricalNamedInputProjector for ConfirmedSwingFactProjector {
fn output_type(&self) -> ValueType {
ValueType::optional(match self.fact {
ConfirmedSwingFactKind::Price => qs_strategy::ScalarType::Price,
_ => qs_strategy::ScalarType::Timestamp,
})
}
fn project(
&self,
context: NamedInputProjectionContext<'_>,
) -> Result<ProjectedNamedInput, NamedInputProjectionError> {
let newest = context
.observations
.iter()
.filter(|observation| observation.source_series().contains(&self.series_id))
.filter_map(|observation| {
observation
.value()
.swing()
.map(|swing| (observation.sequence(), swing))
})
.filter(|(_, swing)| swing.kind() == self.swing_kind)
.max_by_key(|(sequence, _)| *sequence);
let mut retained = self.retained.borrow_mut();
let updated = newest.is_some_and(|(sequence, _)| {
retained
.as_ref()
.is_none_or(|(previous, _)| sequence > *previous)
});
if let Some((sequence, swing)) = newest
&& updated
{
let value = match self.fact {
ConfirmedSwingFactKind::Price => Value::Price(swing.price()),
ConfirmedSwingFactKind::AnchorOpenTime => {
Value::Timestamp(swing.anchor_open_time())
}
ConfirmedSwingFactKind::AnchorCloseTime => {
Value::Timestamp(swing.anchor_close_time())
}
ConfirmedSwingFactKind::ConfirmedAt => Value::Timestamp(swing.confirmed_at()),
};
*retained = Some((sequence, value));
}
Ok(ProjectedNamedInput {
value: retained
.as_ref()
.map(|(_, value)| value.clone())
.unwrap_or(Value::Missing(self.output_type().scalar)),
updated,
})
}
}
pub struct ConfiguredNamedInputBinding {
name: String,
projector: Box<dyn HistoricalNamedInputProjector>,
}
impl ConfiguredNamedInputBinding {
pub fn new(name: impl Into<String>, projector: Box<dyn HistoricalNamedInputProjector>) -> Self {
Self {
name: name.into(),
projector,
}
}
pub fn name(&self) -> &str {
&self.name
}
pub fn output_type(&self) -> ValueType {
self.projector.output_type()
}
}
pub struct ConfiguredHistoricalBindings {
sources: Vec<ConfiguredSourceBinding>,
named_inputs: Vec<ConfiguredNamedInputBinding>,
volume: HistoricalVolumeProjection,
}
impl ConfiguredHistoricalBindings {
pub fn new(
sources: Vec<ConfiguredSourceBinding>,
named_inputs: Vec<ConfiguredNamedInputBinding>,
volume: HistoricalVolumeProjection,
) -> Self {
Self {
sources,
named_inputs,
volume,
}
}
pub fn sources(&self) -> &[ConfiguredSourceBinding] {
&self.sources
}
pub fn named_inputs(&self) -> &[ConfiguredNamedInputBinding] {
&self.named_inputs
}
pub fn volume(&self) -> HistoricalVolumeProjection {
self.volume
}
pub fn into_parts(
self,
) -> (
Vec<ConfiguredSourceBinding>,
Vec<ConfiguredNamedInputBinding>,
HistoricalVolumeProjection,
) {
(self.sources, self.named_inputs, self.volume)
}
}
#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
#[error("{message}")]
pub struct NamedInputProjectionError {
message: String,
}
impl NamedInputProjectionError {
pub fn new(message: impl Into<String>) -> Self {
Self {
message: message.into(),
}
}
}
#[derive(Debug, thiserror::Error)]
pub enum ConfiguredStrategyAdapterBuildError {
#[error("configured source '{source_id}' has no historical binding")]
MissingSourceBinding { source_id: SourceId },
#[error("configured source '{source_id}' is bound more than once")]
DuplicateSourceBinding { source_id: SourceId },
#[error(
"historical series ID '{series_id}' cannot be bound to more than one configured source"
)]
DuplicateSeriesBinding { series_id: SeriesId },
#[error("source '{source_id}' is not declared by the configured strategy")]
UndeclaredSourceBinding { source_id: SourceId },
#[error(
"source '{source_id}' is bound to symbol '{series_symbol}', but the configured strategy primary symbol is '{primary_symbol}'"
)]
SourceSymbolMismatch {
source_id: SourceId,
primary_symbol: String,
series_symbol: String,
},
#[error(
"source '{source_id}' requires lookback {required}, but retained history is {retained}"
)]
RetentionBelowLookback {
source_id: SourceId,
required: usize,
retained: usize,
},
#[error("source '{source_id}' requires lookback {required}, but historical warmup is {warmup}")]
WarmupBelowLookback {
source_id: SourceId,
required: usize,
warmup: usize,
},
#[error("configured named input '{name}' has no projector")]
MissingNamedInputProjector { name: String },
#[error("configured named input '{name}' has more than one projector")]
DuplicateNamedInputProjector { name: String },
#[error("named input '{name}' expects {expected:?}, but its projector returns {actual:?}")]
NamedInputTypeMismatch {
name: String,
expected: ValueType,
actual: ValueType,
},
#[error("named input projector '{name}' is not required by the configured strategy")]
UndeclaredNamedInputProjector { name: String },
#[error(transparent)]
HistoricalRequirements(#[from] StrategyDomainError),
}
#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
pub enum ConfiguredStrategyAdapterPreflightError {
#[error("decision reason capacity {actual} is below configured output capacity {required}")]
DecisionReasonCapacity { actual: usize, required: usize },
#[error("signal capacity {actual} is below configured output capacity {required}")]
SignalCapacity { actual: usize, required: usize },
#[error("journal callback capacity {actual} is below configured output capacity {required}")]
JournalCallbackCapacity { actual: usize, required: usize },
#[error("journal reason capacity {actual} is below configured output capacity {required}")]
JournalReasonCapacity { actual: usize, required: usize },
#[error("journal value capacity {actual} is below configured output capacity {required}")]
JournalValueCapacity { actual: usize, required: usize },
#[error("journal key capacity {actual} is below configured output capacity {required}")]
JournalKeyCapacity { actual: usize, required: usize },
#[error(
"historical trade identity capacity {actual} is below configured identity capacity {required}"
)]
TradeIdentityCapacity { actual: usize, required: usize },
}
#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
pub enum ConfiguredEntryProfileError {
#[error("entry class `{entry_class}` on trade slot `{slot}` has no management-profile route")]
UnroutedEntryClass { entry_class: String, slot: String },
#[error(
"management profile `{profile}` manages the stoploss of trade slot `{slot}`, which the strategy also moves"
)]
StoplossOwnerConflict {
profile: String,
slot: String,
entry_class: Option<String>,
},
}
#[derive(Debug, thiserror::Error)]
pub enum ConfiguredStrategyAdapterError {
#[error("source '{source_id}' produced more than one completed bar at {timestamp}")]
DuplicateSourceUpdate {
source_id: SourceId,
timestamp: NaiveDateTime,
},
#[error("source '{source_id}' requires a tick count but the bar count is unknown")]
MissingTickCount { source_id: SourceId },
#[error("tick count {tick_count} cannot be represented exactly as f64")]
TickCountNotExactlyRepresentable { tick_count: u64 },
#[error("named input '{name}' projection failed: {source}")]
NamedInput {
name: String,
source: NamedInputProjectionError,
},
#[error("named input '{name}' returned a value incompatible with {expected:?}")]
NamedInputValueType { name: String, expected: ValueType },
#[error("trade slot '{slot}' has inconsistent engine state: {reason}")]
TradeSlot { slot: String, reason: String },
#[error("configured command '{command_id}' received an incompatible committed effect")]
IncompatibleCommandEffect { command_id: String },
#[error("configured strategy evaluation failed: {0}")]
Evaluation(#[from] qs_strategy::EvaluationError),
#[error("configured decision mapping failed: {0}")]
Decision(#[from] StrategyDomainError),
#[error("configured note mapping failed: {0}")]
Journal(#[from] StrategyJournalError),
#[error("configured output integer cannot be represented exactly as f64")]
IntegerOutputPrecision,
}
#[derive(Debug, Clone)]
struct CommandRoute {
action: ConfiguredActionKind,
slot: String,
fact_seen: bool,
terminal: Option<CommandTerminalStatus>,
}
pub(crate) struct ConfiguredBoundaryOutput {
pub decision: Option<StrategyDecisionDraft>,
pub journal: Vec<StrategyJournalDraft>,
pub commands: Vec<ConfiguredCommand>,
}
pub(crate) struct BoundaryPositionFacts<'a> {
excursions: &'a BTreeMap<String, CampaignExcursion>,
executor: &'a FutureExecutor,
}
impl<'a> BoundaryPositionFacts<'a> {
pub(crate) fn new(
excursions: &'a BTreeMap<String, CampaignExcursion>,
executor: &'a FutureExecutor,
) -> Self {
Self {
excursions,
executor,
}
}
fn excursion(&self, position_id: &str) -> Option<CampaignExcursion> {
self.excursions
.get(position_id)
.copied()
.filter(|excursion| excursion.observations > 0)
}
fn initial_risk(&self, position_id: &str) -> Option<f64> {
self.executor.open_initial_risk(position_id)
}
}
pub struct BacktestConfiguredStrategyAdapter {
strategy: ConfiguredStrategy,
descriptor: StrategyDescriptor,
requirements: StrategyRequirements,
bindings: ConfiguredHistoricalBindings,
command_routes: BTreeMap<String, CommandRoute>,
evaluation_start: Option<NaiveDateTime>,
first_ready_at: Option<NaiveDateTime>,
}
impl BacktestConfiguredStrategyAdapter {
pub fn new(
strategy: ConfiguredStrategy,
descriptor: StrategyDescriptor,
bindings: ConfiguredHistoricalBindings,
decision_latency_ms: u64,
) -> Result<Self, ConfiguredStrategyAdapterBuildError> {
validate_bindings(&strategy, &bindings)?;
let series = bindings
.sources
.iter()
.map(|binding| binding.series.requirement().clone())
.collect::<Vec<_>>();
let mut instruments = Vec::new();
for requirement in &series {
if !instruments
.iter()
.any(|symbol| symbol == requirement.symbol())
{
instruments.push(requirement.symbol().to_owned());
}
}
let needs_feedback = strategy.input_requirements().needs_command_feedback;
let requirements = StrategyRequirements::new(
instruments,
series,
decision_latency_ms,
true,
needs_feedback,
)?;
Ok(Self {
strategy,
descriptor,
requirements,
bindings,
command_routes: BTreeMap::new(),
evaluation_start: None,
first_ready_at: None,
})
}
pub fn descriptor(&self) -> &StrategyDescriptor {
&self.descriptor
}
pub fn requirements(&self) -> &StrategyRequirements {
&self.requirements
}
pub fn configured_requirements(&self) -> &ConfiguredStrategyRequirements {
self.strategy.input_requirements()
}
pub fn source_bindings(&self) -> &[ConfiguredSourceBinding] {
&self.bindings.sources
}
pub fn series_specs(&self) -> impl ExactSizeIterator<Item = &BarSeriesSpec> {
self.bindings.sources.iter().map(|binding| &binding.series)
}
pub fn first_ready_at(&self) -> Option<NaiveDateTime> {
self.first_ready_at
}
pub fn set_evaluation_start(&mut self, evaluation_start: Option<NaiveDateTime>) {
self.evaluation_start = evaluation_start;
}
pub fn configured_strategy(&self) -> &ConfiguredStrategy {
&self.strategy
}
pub fn into_configured_strategy(self) -> ConfiguredStrategy {
self.strategy
}
pub fn preflight_entry_profiles(
&self,
profiles: &PreparedEntryProfiles,
) -> Result<(), ConfiguredEntryProfileError> {
let requirements = self.strategy.input_requirements();
for entry in &requirements.entries {
let profile = match entry.entry_class.as_ref() {
Some(entry_class) => Some(profiles.routes().get(entry_class).ok_or_else(|| {
ConfiguredEntryProfileError::UnroutedEntryClass {
entry_class: entry_class.clone(),
slot: entry.slot.clone(),
}
})?),
None => profiles.default_profile(),
};
if let Some(profile) = profile
&& profile_manages_stoploss(profile)
&& requirements.stop_managed_slots.contains(&entry.slot)
{
return Err(ConfiguredEntryProfileError::StoplossOwnerConflict {
profile: profile.name.clone(),
slot: entry.slot.clone(),
entry_class: entry.entry_class.clone(),
});
}
}
Ok(())
}
pub fn preflight(
&self,
retention: StrategyRetentionLimits,
research: StrategyResearchLimits,
) -> Result<(), ConfiguredStrategyAdapterPreflightError> {
if retention.max_reason_bytes() < MAX_TEXT_BYTES {
return Err(
ConfiguredStrategyAdapterPreflightError::DecisionReasonCapacity {
actual: retention.max_reason_bytes(),
required: MAX_TEXT_BYTES,
},
);
}
if retention.max_signals_per_callback() < MAX_OUTPUT_COMMANDS {
return Err(ConfiguredStrategyAdapterPreflightError::SignalCapacity {
actual: retention.max_signals_per_callback(),
required: MAX_OUTPUT_COMMANDS,
});
}
if research.max_journal_per_callback() < MAX_OUTPUT_NOTES {
return Err(
ConfiguredStrategyAdapterPreflightError::JournalCallbackCapacity {
actual: research.max_journal_per_callback(),
required: MAX_OUTPUT_NOTES,
},
);
}
if research.max_reason_bytes() < MAX_TEXT_BYTES {
return Err(
ConfiguredStrategyAdapterPreflightError::JournalReasonCapacity {
actual: research.max_reason_bytes(),
required: MAX_TEXT_BYTES,
},
);
}
if research.max_values_per_record() < MAX_NAMED_VALUES {
return Err(
ConfiguredStrategyAdapterPreflightError::JournalValueCapacity {
actual: research.max_values_per_record(),
required: MAX_NAMED_VALUES,
},
);
}
if research.max_value_key_bytes() < MAX_ID_BYTES {
return Err(
ConfiguredStrategyAdapterPreflightError::JournalKeyCapacity {
actual: research.max_value_key_bytes(),
required: MAX_ID_BYTES,
},
);
}
if super::MAX_TRADE_ID_BYTES < MAX_GENERATED_ID_BYTES {
return Err(
ConfiguredStrategyAdapterPreflightError::TradeIdentityCapacity {
actual: super::MAX_TRADE_ID_BYTES,
required: MAX_GENERATED_ID_BYTES,
},
);
}
Ok(())
}
#[allow(clippy::too_many_arguments)]
pub(crate) fn evaluate_boundary(
&mut self,
observed_through: NaiveDateTime,
ready: bool,
closed_bars: &[ClosedBar],
observations: &[StrategyObservation],
series: &dyn HistoricalSeriesView,
observation_history: &dyn HistoricalObservationView,
engine: &TradeEngine,
positions: &BoundaryPositionFacts<'_>,
feedback_events: &[StrategyFeedbackEvent],
retention: StrategyRetentionLimits,
research: StrategyResearchLimits,
) -> Result<ConfiguredBoundaryOutput, ConfiguredStrategyAdapterError> {
let ready = ready
&& self
.evaluation_start
.is_none_or(|start| observed_through >= start);
if ready && self.first_ready_at.is_none() {
self.first_ready_at = Some(observed_through);
}
let feedback = self.project_feedback(feedback_events)?;
let input = StrategyInput {
time: observed_through,
ready,
completed_bars: self.project_bars(observed_through, closed_bars)?,
values: self.project_named_inputs(NamedInputProjectionContext {
observed_through,
closed_bars,
observations,
series,
observation_history,
})?,
trade_slots: self.project_trade_slots(engine, positions)?,
feedback,
};
let output = self.strategy.evaluate(&input)?;
for command in &output.commands {
self.command_routes.insert(
command.command_id.clone(),
CommandRoute {
action: command.action_kind,
slot: command.trade_slot.clone(),
fact_seen: false,
terminal: None,
},
);
}
let emitted_signals = output
.commands
.iter()
.map(|command| command.signal.clone())
.collect::<Vec<_>>();
let decision = output
.decision
.map(|decision| map_decision(decision, emitted_signals, retention))
.transpose()?;
let journal = output
.notes
.into_iter()
.map(|note| map_note(note, self.strategy.primary_symbol(), research))
.collect::<Result<Vec<_>, _>>()?;
Ok(ConfiguredBoundaryOutput {
decision,
journal,
commands: output.commands,
})
}
fn project_bars(
&self,
observed_through: NaiveDateTime,
closed_bars: &[ClosedBar],
) -> Result<Vec<qs_strategy::CompletedBarUpdate>, ConfiguredStrategyAdapterError> {
self.strategy
.input_requirements()
.completed_bars
.iter()
.filter_map(|requirement| {
let binding = self
.bindings
.sources
.iter()
.find(|binding| binding.source == requirement.source)
.expect("bindings were validated at construction");
let mut matching = closed_bars
.iter()
.filter(|bar| bar.series_id() == binding.series_id());
let bar = matching.next()?;
Some(if matching.next().is_some() {
Err(ConfiguredStrategyAdapterError::DuplicateSourceUpdate {
source_id: requirement.source.clone(),
timestamp: observed_through,
})
} else {
Ok(qs_strategy::CompletedBarUpdate {
source: requirement.source.clone(),
bar: qs_strategy::CompletedBar {
open: bar.open(),
high: bar.high(),
low: bar.low(),
close: bar.close(),
volume: match self.bindings.volume {
HistoricalVolumeProjection::TickCountExact => {
let tick_count = match bar.tick_count() {
Some(count) => count,
None => return Some(Err(ConfiguredStrategyAdapterError::MissingTickCount { source_id: requirement.source.clone() })),
};
if tick_count > MAX_EXACT_F64_INTEGER {
return Some(Err(
ConfiguredStrategyAdapterError::TickCountNotExactlyRepresentable { tick_count },
));
}
Some(tick_count as f64)
}
HistoricalVolumeProjection::OptionalTickCount => bar.tick_count().map(|count| count as f64)
},
},
})
})
})
.collect()
}
fn project_named_inputs(
&self,
context: NamedInputProjectionContext<'_>,
) -> Result<Vec<NamedValue>, ConfiguredStrategyAdapterError> {
self.strategy
.input_requirements()
.named_inputs
.iter()
.map(|requirement| {
let binding = self
.bindings
.named_inputs
.iter()
.find(|binding| binding.name == requirement.name)
.expect("named input bindings were validated at construction");
let projected = binding.projector.project(context).map_err(|source| {
ConfiguredStrategyAdapterError::NamedInput {
name: requirement.name.clone(),
source,
}
})?;
if !value_matches_type(&projected.value, requirement.value_type) {
return Err(ConfiguredStrategyAdapterError::NamedInputValueType {
name: requirement.name.clone(),
expected: requirement.value_type,
});
}
Ok(NamedValue {
name: requirement.name.clone(),
value: projected.value,
updated: projected.updated,
})
})
.collect()
}
fn project_trade_slots(
&self,
engine: &TradeEngine,
positions: &BoundaryPositionFacts<'_>,
) -> Result<Vec<TradeSlotFacts>, ConfiguredStrategyAdapterError> {
self.strategy
.input_requirements()
.trade_slots
.iter()
.map(|slot| {
let position = self
.strategy
.trade_id_for_slot(slot)
.and_then(|trade_id| engine.manager.id_by_trade_id(trade_id))
.and_then(|position_id| {
engine
.get_position(&position_id)
.map(|position| (position_id, position))
});
let state = match position {
None => TradeSlotState::Vacant,
Some((position_id, position)) => match position.data.status {
PositionStatus::Pending => TradeSlotState::Pending {
side: position.data.side,
requested_price: position.data.pending_price,
stoploss: position.current_stoploss(),
},
PositionStatus::Open => {
let opened_at = position.data.open_ts.ok_or_else(|| {
ConfiguredStrategyAdapterError::TradeSlot {
slot: slot.clone(),
reason: "open position has no entry fill time".into(),
}
})?;
let excursion = positions.excursion(&position_id);
TradeSlotState::Open {
side: position.data.side,
entry_price: position.data.average_entry(),
remaining_size: position.data.remaining_size(),
stoploss: position.current_stoploss(),
opened_at,
favorable_excursion: excursion.map(|excursion| excursion.mfe),
adverse_excursion: excursion.map(|excursion| excursion.mae),
initial_risk: positions.initial_risk(&position_id),
}
}
PositionStatus::Closed | PositionStatus::Cancelled => {
TradeSlotState::Vacant
}
},
};
Ok(TradeSlotFacts {
slot: slot.clone(),
state,
})
})
.collect()
}
fn project_feedback(
&mut self,
events: &[StrategyFeedbackEvent],
) -> Result<Vec<CommandFeedback>, ConfiguredStrategyAdapterError> {
project_command_feedback(&mut self.command_routes, events)
}
pub(crate) fn finalize_feedback(
&mut self,
events: &[StrategyFeedbackEvent],
) -> Result<(), ConfiguredStrategyAdapterError> {
let feedback = self.project_feedback(events)?;
self.strategy.finalize_command_feedback(&feedback)?;
self.command_routes.clear();
Ok(())
}
}
fn project_command_feedback(
routes: &mut BTreeMap<String, CommandRoute>,
events: &[StrategyFeedbackEvent],
) -> Result<Vec<CommandFeedback>, ConfiguredStrategyAdapterError> {
let mut projected = Vec::new();
for event in events {
let Some(command_id) = event.action_id() else {
continue;
};
let Some(route) = routes.get_mut(command_id) else {
continue;
};
match event {
StrategyFeedbackEvent::Effect { effect, .. } => {
if let Some(fact) = map_effect(route.action, effect.effect()).map_err(|()| {
ConfiguredStrategyAdapterError::IncompatibleCommandEffect {
command_id: command_id.to_owned(),
}
})? {
route.fact_seen = true;
projected.push(CommandFeedback::Fact {
command_id: command_id.to_owned(),
fact,
});
}
}
StrategyFeedbackEvent::Disposition(disposition) => {
let status = match disposition.status {
ActionDispositionStatus::Applied => CommandTerminalStatus::Applied,
ActionDispositionStatus::Skipped => CommandTerminalStatus::Skipped,
ActionDispositionStatus::Rejected => CommandTerminalStatus::Rejected,
ActionDispositionStatus::Failed => CommandTerminalStatus::Failed,
};
route.terminal = Some(status);
projected.push(CommandFeedback::Terminal {
command_id: command_id.to_owned(),
status,
reason: disposition.reason.clone(),
});
}
}
let completed = route
.terminal
.is_some_and(|status| status != CommandTerminalStatus::Applied)
|| (route.terminal == Some(CommandTerminalStatus::Applied) && route.fact_seen);
if completed {
let completed_route = routes
.remove(command_id)
.expect("completed route remains registered");
if completed_route.action == ConfiguredActionKind::CancelPending
&& completed_route.terminal == Some(CommandTerminalStatus::Applied)
{
routes.retain(|_, route| {
!(route.action == ConfiguredActionKind::Entry
&& route.slot == completed_route.slot)
});
}
}
}
Ok(projected)
}
fn validate_bindings(
strategy: &ConfiguredStrategy,
bindings: &ConfiguredHistoricalBindings,
) -> Result<(), ConfiguredStrategyAdapterBuildError> {
let declared = strategy.declared_sources().iter().collect::<BTreeSet<_>>();
let mut sources = BTreeSet::new();
let mut series = BTreeSet::new();
for binding in &bindings.sources {
if !declared.contains(&binding.source) {
return Err(
ConfiguredStrategyAdapterBuildError::UndeclaredSourceBinding {
source_id: binding.source.clone(),
},
);
}
if !sources.insert(binding.source.clone()) {
return Err(
ConfiguredStrategyAdapterBuildError::DuplicateSourceBinding {
source_id: binding.source.clone(),
},
);
}
if !series.insert(binding.series_id().clone()) {
return Err(
ConfiguredStrategyAdapterBuildError::DuplicateSeriesBinding {
series_id: binding.series_id().clone(),
},
);
}
let series_symbol = binding.series.requirement().symbol();
if series_symbol != strategy.primary_symbol() {
return Err(ConfiguredStrategyAdapterBuildError::SourceSymbolMismatch {
source_id: binding.source.clone(),
primary_symbol: strategy.primary_symbol().to_owned(),
series_symbol: series_symbol.to_owned(),
});
}
}
for source in strategy.declared_sources() {
if !sources.contains(source) {
return Err(ConfiguredStrategyAdapterBuildError::MissingSourceBinding {
source_id: source.clone(),
});
}
}
for requirement in &strategy.input_requirements().completed_bars {
let binding = bindings
.sources
.iter()
.find(|binding| binding.source == requirement.source)
.expect("every declared source was checked above");
if binding.series.retained_bars() < requirement.required_lookback {
return Err(
ConfiguredStrategyAdapterBuildError::RetentionBelowLookback {
source_id: requirement.source.clone(),
required: requirement.required_lookback,
retained: binding.series.retained_bars(),
},
);
}
let warmup = binding.series.requirement().warmup().required_bars();
if warmup < requirement.required_lookback {
return Err(ConfiguredStrategyAdapterBuildError::WarmupBelowLookback {
source_id: requirement.source.clone(),
required: requirement.required_lookback,
warmup,
});
}
}
let mut names = BTreeSet::new();
for binding in &bindings.named_inputs {
if !names.insert(binding.name.clone()) {
return Err(
ConfiguredStrategyAdapterBuildError::DuplicateNamedInputProjector {
name: binding.name.clone(),
},
);
}
let Some(requirement) = strategy
.input_requirements()
.named_inputs
.iter()
.find(|requirement| requirement.name == binding.name)
else {
return Err(
ConfiguredStrategyAdapterBuildError::UndeclaredNamedInputProjector {
name: binding.name.clone(),
},
);
};
let actual = binding.output_type();
if actual != requirement.value_type {
return Err(
ConfiguredStrategyAdapterBuildError::NamedInputTypeMismatch {
name: binding.name.clone(),
expected: requirement.value_type,
actual,
},
);
}
}
for requirement in &strategy.input_requirements().named_inputs {
if !names.contains(&requirement.name) {
return Err(
ConfiguredStrategyAdapterBuildError::MissingNamedInputProjector {
name: requirement.name.clone(),
},
);
}
}
Ok(())
}
fn value_matches_type(value: &Value, expected: ValueType) -> bool {
if value.is_missing() {
return expected.optional && value.scalar_type() == expected.scalar;
}
if value.scalar_type() != expected.scalar {
return false;
}
match value {
Value::Number(value)
| Value::Price(value)
| Value::Ratio(value)
| Value::Percent(value)
| Value::PricePerObservation(value)
| Value::PricePerObservationSquared(value)
| Value::RatioPerObservation(value)
| Value::RatioPerObservationSquared(value)
| Value::LogReturn(value)
| Value::LogReturnVariance(value) => value.is_finite(),
Value::Text(value) => !value.is_empty() && value.len() <= MAX_TEXT_BYTES,
_ => true,
}
}
fn profile_manages_stoploss(profile: &ManagementProfile) -> bool {
!matches!(profile.stoploss_mode, StoplossMode::FromSignal)
|| profile.rules.iter().any(|rule| {
matches!(
rule,
RuleConfigDef::FixedStoploss { .. }
| RuleConfigDef::TrailingStop { .. }
| RuleConfigDef::BreakevenWhen { .. }
| RuleConfigDef::BreakevenWhenOffset { .. }
| RuleConfigDef::BreakevenAfterTargets { .. }
)
})
}
fn map_effect(action: ConfiguredActionKind, effect: &Effect) -> Result<Option<CommandFact>, ()> {
let mapped = match effect {
Effect::PositionOpened { .. } => {
Some((ConfiguredActionKind::Entry, CommandFact::EntryFilled))
}
Effect::PositionClosed { .. } => {
Some((ConfiguredActionKind::Close, CommandFact::PositionClosed))
}
Effect::PartialClose { .. } => Some((
ConfiguredActionKind::ClosePartial,
CommandFact::PositionReduced,
)),
Effect::StoplossModified { .. } => match action {
ConfiguredActionKind::MoveStoplossToEntry | ConfiguredActionKind::ModifyStoploss => {
return Ok(Some(CommandFact::StoplossModified));
}
_ => return Err(()),
},
Effect::OrderCancelled { .. } => Some((
ConfiguredActionKind::CancelPending,
CommandFact::PendingCancelled,
)),
Effect::OrderPlaced { .. }
| Effect::StoplossRemoved { .. }
| Effect::ScaledIn { .. }
| Effect::RuleTriggered { .. } => None,
};
match mapped {
Some((expected, fact)) if expected == action => Ok(Some(fact)),
Some(_) => Err(()),
None => Ok(None),
}
}
fn map_decision(
decision: qs_strategy::Decision,
emitted_signals: Vec<qs_core::RawSignal>,
limits: StrategyRetentionLimits,
) -> Result<StrategyDecisionDraft, StrategyDomainError> {
let kind = match decision.kind {
DecisionKind::Entry => StrategyDecisionKind::Entry,
DecisionKind::Management => StrategyDecisionKind::Management,
DecisionKind::Exit => StrategyDecisionKind::Exit,
DecisionKind::Observation => StrategyDecisionKind::Annotation,
};
StrategyDecisionDraft::new(
kind,
decision.reason,
decision.related_trade.map(|trade| trade.trade_id),
emitted_signals,
limits,
)
}
fn map_note(
note: qs_strategy::Note,
symbol: &str,
limits: StrategyResearchLimits,
) -> Result<StrategyJournalDraft, ConfiguredStrategyAdapterError> {
let kind = match note.kind {
NoteKind::Observation | NoteKind::Risk => JournalKind::DecisionContext,
NoteKind::Execution | NoteKind::Lifecycle => JournalKind::OutcomeReview,
};
let mut values = BTreeMap::new();
for output in note.values {
let value = match output.value {
OutputScalar::Bool(value) => f64::from(value),
OutputScalar::Integer(value) => {
if value.unsigned_abs() > MAX_EXACT_F64_INTEGER {
return Err(ConfiguredStrategyAdapterError::IntegerOutputPrecision);
}
value as f64
}
OutputScalar::Number(value)
| OutputScalar::Price(value)
| OutputScalar::Ratio(value)
| OutputScalar::Percent(value)
| OutputScalar::PricePerObservation(value)
| OutputScalar::PricePerObservationSquared(value)
| OutputScalar::RatioPerObservation(value)
| OutputScalar::RatioPerObservationSquared(value)
| OutputScalar::LogReturn(value)
| OutputScalar::LogReturnVariance(value) => value,
};
values.insert(output.name, value);
}
Ok(StrategyJournalDraft::new(
kind,
symbol,
note.related_trade.map(|trade| trade.trade_id),
note.reason,
None,
values,
limits,
)?)
}
#[cfg(test)]
mod tests {
use super::*;
use crate::ledger::ActionDisposition;
use qs_core::types::FutureEffect;
#[test]
fn applied_terminal_before_effect_completes_command_correlation() {
let command_id = "opaque-command".to_owned();
let mut routes = BTreeMap::from([(
command_id.clone(),
CommandRoute {
action: ConfiguredActionKind::Entry,
slot: "primary".into(),
fact_seen: false,
terminal: None,
},
)]);
let terminal = project_command_feedback(
&mut routes,
&[StrategyFeedbackEvent::Disposition(
ActionDisposition::applied(command_id.clone()),
)],
)
.unwrap();
assert_eq!(
terminal,
vec![CommandFeedback::Terminal {
command_id: command_id.clone(),
status: CommandTerminalStatus::Applied,
reason: None,
}]
);
assert!(routes.contains_key(&command_id));
let fact = project_command_feedback(
&mut routes,
&[StrategyFeedbackEvent::Effect {
action_id: Some(command_id.clone()),
effect: FutureEffect::plain(Effect::PositionOpened {
id: "position-1".into(),
}),
}],
)
.unwrap();
assert_eq!(
fact,
vec![CommandFeedback::Fact {
command_id,
fact: CommandFact::EntryFilled,
}]
);
assert!(routes.is_empty());
}
}