|
|
|
@@ -2,6 +2,7 @@
|
|
|
|
|
//! prices are intentionally absent; the existing broker owns those decisions.
|
|
|
|
|
|
|
|
|
|
use std::collections::{BTreeMap, BTreeSet};
|
|
|
|
|
use std::sync::{Arc, Mutex, OnceLock, Weak};
|
|
|
|
|
|
|
|
|
|
use chrono::{DateTime, FixedOffset, NaiveDate, NaiveDateTime, NaiveTime, Utc};
|
|
|
|
|
use serde::{Deserialize, Serialize};
|
|
|
|
@@ -12,6 +13,67 @@ use crate::portfolio::PortfolioState;
|
|
|
|
|
|
|
|
|
|
pub const SIGNAL_BOOK_SCHEMA: &str = "fidc.signal-book/v2";
|
|
|
|
|
|
|
|
|
|
#[derive(Debug, Clone, PartialEq, Eq, Deserialize, Serialize)]
|
|
|
|
|
#[serde(rename_all = "camelCase", deny_unknown_fields)]
|
|
|
|
|
pub struct SignalBookReference {
|
|
|
|
|
pub book_id: String,
|
|
|
|
|
pub version_sha256: String,
|
|
|
|
|
pub artifact_sha256: String,
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
impl SignalBookReference {
|
|
|
|
|
pub fn validate(&self) -> Result<(), String> {
|
|
|
|
|
if !valid_sha(&self.version_sha256) || !valid_sha(&self.artifact_sha256)
|
|
|
|
|
|| self.book_id != format!("signal_book_{}",self.version_sha256)
|
|
|
|
|
{ return Err("signal_book_reference_invalid".into()); }
|
|
|
|
|
Ok(())
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
#[derive(Default)]
|
|
|
|
|
struct SignalCache {
|
|
|
|
|
entries: BTreeMap<String,Weak<ValidatedSignalBook>>,
|
|
|
|
|
retained: std::collections::VecDeque<(String,Arc<ValidatedSignalBook>,usize)>,
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
fn signal_cache() -> &'static Mutex<SignalCache> {
|
|
|
|
|
static CACHE: OnceLock<Mutex<SignalCache>> = OnceLock::new();
|
|
|
|
|
CACHE.get_or_init(||Mutex::new(SignalCache::default()))
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
pub fn cached_signal_book(reference: &SignalBookReference) -> Result<Option<Arc<ValidatedSignalBook>>,String> {
|
|
|
|
|
reference.validate()?;
|
|
|
|
|
let cache=signal_cache().lock().map_err(|_|"signal_cache_lock_failed")?;
|
|
|
|
|
let book=cache.entries.get(&reference.artifact_sha256).and_then(Weak::upgrade);
|
|
|
|
|
if book.as_ref().is_some_and(|book|book.version_sha256()!=reference.version_sha256) {
|
|
|
|
|
return Err("signal_book_cached_version_mismatch".into());
|
|
|
|
|
}
|
|
|
|
|
Ok(book)
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
pub fn register_signal_book(reference: &SignalBookReference, body: &[u8]) -> Result<Arc<ValidatedSignalBook>,String> {
|
|
|
|
|
use sha2::{Digest,Sha256};
|
|
|
|
|
reference.validate()?;
|
|
|
|
|
if body.len()>64*1024*1024 || format!("{:x}",Sha256::digest(body))!=reference.artifact_sha256 {
|
|
|
|
|
return Err("signal_book_artifact_hash_or_size_invalid".into());
|
|
|
|
|
}
|
|
|
|
|
let raw:SignalBook=serde_json::from_slice(body).map_err(|error|format!("signal_book_decode_failed: {error}"))?;
|
|
|
|
|
if raw.version_sha256!=reference.version_sha256 { return Err("signal_book_version_mismatch".into()); }
|
|
|
|
|
let book=Arc::new(raw.validate()?);
|
|
|
|
|
let mut cache=signal_cache().lock().map_err(|_|"signal_cache_lock_failed")?;
|
|
|
|
|
cache.entries.retain(|_,value|value.strong_count()>0);
|
|
|
|
|
if let Some(existing)=cache.entries.get(&reference.artifact_sha256).and_then(Weak::upgrade) { return Ok(existing); }
|
|
|
|
|
cache.entries.insert(reference.artifact_sha256.clone(),Arc::downgrade(&book));
|
|
|
|
|
let estimated=body.len().saturating_mul(4);
|
|
|
|
|
if estimated<=128*1024*1024 {
|
|
|
|
|
cache.retained.push_back((reference.artifact_sha256.clone(),book.clone(),estimated));
|
|
|
|
|
while cache.retained.len()>4 || cache.retained.iter().map(|entry|entry.2).sum::<usize>()>128*1024*1024 {
|
|
|
|
|
cache.retained.pop_front();
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
Ok(book)
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize, Serialize)]
|
|
|
|
|
#[serde(rename_all = "snake_case")]
|
|
|
|
|
pub enum SignalProvenance {
|
|
|
|
@@ -92,11 +154,33 @@ impl SignalBook {
|
|
|
|
|
pub fn content_sha256(&self) -> Result<String, String> {
|
|
|
|
|
let mut value=serde_json::to_value(self).map_err(|error|error.to_string())?;
|
|
|
|
|
value.as_object_mut().ok_or("signal_book_object_required")?.remove("versionSha256");
|
|
|
|
|
for snapshot in value["snapshots"].as_array_mut().ok_or("signal_snapshots_required")? {
|
|
|
|
|
let object=snapshot.as_object_mut().ok_or("signal_snapshot_object_required")?;
|
|
|
|
|
object.remove("generatedAt"); object.remove("publishedAt");
|
|
|
|
|
value["knowledgeCutoff"]=self.knowledge_cutoff.map(|at|serde_json::json!(at.timestamp_micros())).unwrap_or(serde_json::Value::Null);
|
|
|
|
|
value["expectedDecisions"]=serde_json::json!(self.expected_decisions.iter().map(DateTime::timestamp_micros).collect::<Vec<_>>());
|
|
|
|
|
for (raw,snapshot) in value["snapshots"].as_array_mut().ok_or("signal_snapshots_required")?.iter_mut().zip(&self.snapshots) {
|
|
|
|
|
let object=raw.as_object_mut().ok_or("signal_snapshot_required")?;
|
|
|
|
|
object.remove("generatedAt");
|
|
|
|
|
object.remove("publishedAt");
|
|
|
|
|
for (key,at) in [("signalAt",snapshot.signal_at),("decisionAt",snapshot.decision_at),
|
|
|
|
|
("inputAsOf",snapshot.input_as_of),("inputAvailableAt",snapshot.input_available_at)] {
|
|
|
|
|
object.insert(key.into(),serde_json::json!(at.timestamp_micros()));
|
|
|
|
|
}
|
|
|
|
|
for (raw,action) in object.get_mut("actions").and_then(serde_json::Value::as_array_mut).ok_or("signal_actions_required")?.iter_mut().zip(&snapshot.actions) {
|
|
|
|
|
match action {
|
|
|
|
|
SignalAction::TargetWeight{weight,..}=>raw["weight"]=serde_json::json!(format!("{:016x}",weight.to_bits())),
|
|
|
|
|
SignalAction::Reduce{remaining_ratio,..}=>raw["remaining_ratio"]=serde_json::json!(format!("{:016x}",remaining_ratio.to_bits())),
|
|
|
|
|
_=>{}
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
let raw=serde_json::to_vec(&value).map_err(|error|error.to_string())?;
|
|
|
|
|
fn sorted(value:serde_json::Value)->serde_json::Value {
|
|
|
|
|
match value {
|
|
|
|
|
serde_json::Value::Object(map)=>serde_json::Value::Object(map.into_iter().map(|(key,value)|(key,sorted(value)))
|
|
|
|
|
.collect::<BTreeMap<_,_>>().into_iter().collect()),
|
|
|
|
|
serde_json::Value::Array(rows)=>serde_json::Value::Array(rows.into_iter().map(sorted).collect()),
|
|
|
|
|
other=>other,
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
let raw=serde_json::to_vec(&sorted(value)).map_err(|error|error.to_string())?;
|
|
|
|
|
Ok(format!("{:x}",Sha256::digest(raw)))
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
@@ -106,10 +190,9 @@ impl SignalBook {
|
|
|
|
|
{
|
|
|
|
|
return Err("signal_book_identity_invalid".into());
|
|
|
|
|
}
|
|
|
|
|
if self.model_sha256.is_some()!=self.knowledge_cutoff.is_some()
|
|
|
|
|
|| self.model_sha256.as_ref().is_some_and(|value|!valid_sha(value)) {
|
|
|
|
|
return Err("signal_model_training_identity_incomplete".into());
|
|
|
|
|
}
|
|
|
|
|
if self.model_sha256.as_ref().is_some_and(|value| !valid_sha(value))
|
|
|
|
|
|| self.model_sha256.is_some() != self.knowledge_cutoff.is_some()
|
|
|
|
|
{ return Err("signal_model_training_identity_incomplete".into()); }
|
|
|
|
|
if self.expected_decisions.is_empty() || self.expected_decisions.len() > 100_000
|
|
|
|
|
|| self.expected_decisions.len() != self.snapshots.len()
|
|
|
|
|
{
|
|
|
|
@@ -119,16 +202,19 @@ impl SignalBook {
|
|
|
|
|
let mut previous = None;
|
|
|
|
|
let mut total_actions = 0usize;
|
|
|
|
|
for (number, (expected, snapshot)) in self.expected_decisions.iter().zip(&self.snapshots).enumerate() {
|
|
|
|
|
if [*expected,snapshot.signal_at,snapshot.input_as_of,snapshot.input_available_at,snapshot.generated_at,snapshot.published_at]
|
|
|
|
|
.iter().any(|at|at.timestamp_subsec_nanos()%1000!=0) || self.knowledge_cutoff.is_some_and(|at|at.timestamp_subsec_nanos()%1000!=0) {
|
|
|
|
|
return Err("signal_timestamp_requires_microsecond_precision".into());
|
|
|
|
|
}
|
|
|
|
|
if snapshot.decision_at != *expected || previous.is_some_and(|value| value >= *expected) {
|
|
|
|
|
return Err("signal_book_decisions_duplicate_or_unordered".into());
|
|
|
|
|
}
|
|
|
|
|
previous = Some(*expected);
|
|
|
|
|
if self.knowledge_cutoff.is_some_and(|cutoff|cutoff>snapshot.signal_at)
|
|
|
|
|
|| snapshot.input_as_of > snapshot.input_available_at
|
|
|
|
|
|| snapshot.input_available_at > snapshot.signal_at || snapshot.signal_at > *expected
|
|
|
|
|
if self.knowledge_cutoff.is_some_and(|cutoff| cutoff > snapshot.signal_at) || snapshot.signal_at > *expected
|
|
|
|
|
|| snapshot.input_available_at > snapshot.signal_at || snapshot.input_as_of > snapshot.input_available_at
|
|
|
|
|
|| snapshot.published_at < snapshot.generated_at || !valid_sha(&snapshot.input_sha256)
|
|
|
|
|
|| snapshot.generated_at < snapshot.input_available_at
|
|
|
|
|
|| self.knowledge_cutoff.is_some_and(|cutoff|snapshot.generated_at<cutoff)
|
|
|
|
|
|| self.knowledge_cutoff.is_some_and(|cutoff| snapshot.generated_at < cutoff)
|
|
|
|
|
{
|
|
|
|
|
return Err("signal_book_future_or_invalid_input".into());
|
|
|
|
|
}
|
|
|
|
@@ -201,7 +287,11 @@ impl ValidatedSignalBook {
|
|
|
|
|
|
|
|
|
|
pub fn snapshot_for(&self, ctx: &StrategyContext<'_>) -> Result<&SignalSnapshot, String> {
|
|
|
|
|
let snapshot = self.snapshot_at(ctx.execution_date, ctx.current_time(), ctx.is_lagged_execution())?;
|
|
|
|
|
if ctx.is_lagged_execution() && shanghai(snapshot.input_as_of).date() > ctx.decision_date {
|
|
|
|
|
let logical_clock=ctx.current_datetime().filter(|at|at.date()==ctx.decision_date)
|
|
|
|
|
.unwrap_or(ctx.decision_date.and_hms_opt(15,0,0).expect("completed decision session"));
|
|
|
|
|
let lagged_daily=ctx.is_lagged_execution() && self.book.frequency==SignalFrequency::Daily;
|
|
|
|
|
if (lagged_daily && shanghai(snapshot.input_as_of).date()>ctx.decision_date)
|
|
|
|
|
|| (!lagged_daily && shanghai(snapshot.signal_at)>logical_clock) {
|
|
|
|
|
return Err("next_open_signal_contains_execution_session_inputs".into());
|
|
|
|
|
}
|
|
|
|
|
Ok(snapshot)
|
|
|
|
@@ -284,11 +374,12 @@ mod tests {
|
|
|
|
|
let source: DateTime<Utc> = "2025-01-06T15:00:00+08:00".parse().unwrap();
|
|
|
|
|
seal(SignalBook {
|
|
|
|
|
schema: SIGNAL_BOOK_SCHEMA.into(), version_sha256: "a".repeat(64), generator_sha256: "b".repeat(64),
|
|
|
|
|
model_sha256:Some("d".repeat(64)),
|
|
|
|
|
model_sha256: Some("d".repeat(64)),
|
|
|
|
|
knowledge_cutoff: Some("2024-12-31T15:00:00+08:00".parse().unwrap()),
|
|
|
|
|
provenance: SignalProvenance::Reconstructed, frequency: SignalFrequency::Daily,
|
|
|
|
|
expected_decisions: vec![decision], snapshots: vec![SignalSnapshot {
|
|
|
|
|
signal_at:source, decision_at: decision, input_as_of: source, input_available_at: source,
|
|
|
|
|
signal_at: source,
|
|
|
|
|
decision_at: decision, input_as_of: source, input_available_at: source,
|
|
|
|
|
generated_at: decision + Duration::days(10), published_at: decision + Duration::days(10),
|
|
|
|
|
input_sha256: "c".repeat(64), complete_targets: true,
|
|
|
|
|
actions: vec![SignalAction::TargetWeight { symbol: "000001.SZ".into(), weight: 0.5 }],
|
|
|
|
|