diff --git a/examples/sma.myc b/examples/sma.myc index 617c17d..ca3acff 100644 --- a/examples/sma.myc +++ b/examples/sma.myc @@ -1,12 +1,10 @@ #use rtl (do - (def sma20 (SMA 20)) - - (def n 100) - (while (> n 0) - (do - (print "val=" n " sma20=" (sma20 n)) - (assign n (- n 1)) - )) + (def src (create-random-ohlc 42 200)) + + (macro printx [s] `(fn [x] (print ~s x))) + + (pipe [(.close src)] (printx "close=")) + (pipe [(pipe [(.close src)] (SMA 20))] (printx "sma=")) ) diff --git a/examples/streams.myc b/examples/streams.myc new file mode 100644 index 0000000..0d8bb76 --- /dev/null +++ b/examples/streams.myc @@ -0,0 +1,10 @@ +(do +(def stream (create-random-ohlc 42 200)) + +(macro prt [m] `(pipe (~m stream) (fn [s] (print ~m "=" s)))) + +(prt .open) +(prt .high) +(prt .low) +(prt .close) +) \ No newline at end of file diff --git a/src/ast/compiler/type_checker.rs b/src/ast/compiler/type_checker.rs index 71a3d07..686006d 100644 --- a/src/ast/compiler/type_checker.rs +++ b/src/ast/compiler/type_checker.rs @@ -422,9 +422,11 @@ impl TypeChecker { let mut arg_types = Vec::with_capacity(inputs.len()); for input in inputs { let typed_input = self.check_node(input, ctx, diag); - // Unwrap Series(T) to T for the lambda argument + // Unwrap Series(T) or Stream(T) to T for the lambda argument let arg_ty = if let StaticType::Series(inner) = &typed_input.ty { *inner.clone() + } else if let StaticType::Stream(inner) = &typed_input.ty { + *inner.clone() } else { StaticType::Any }; diff --git a/src/ast/rtl/streams.rs b/src/ast/rtl/streams.rs index e0a2d1f..7c48c64 100644 --- a/src/ast/rtl/streams.rs +++ b/src/ast/rtl/streams.rs @@ -416,6 +416,34 @@ pub fn build_pipeline_node( }) } +pub fn build_map_stream(input: Rc, field: Keyword) -> StreamNode { + let executor: Box = Box::new(move |args: &[Value]| -> Value { + let val = &args[0]; + if let Value::Record(layout, values) = val + && let Some(idx) = layout.index_of(field) + { + return values[idx].clone(); + } + Value::Void // In streams, Void acts as a filter + }); + + let pipe = Rc::new(RefCell::new(PipeStream::new( + format!("map:{}", field.name()), + 1, + Some(executor), + ))); + + let adapter = Rc::new(RefCell::new(SourceAdapter { + target: pipe.clone(), + target_index: 0, + })); + input.add_observer(adapter); + + StreamNode { + inner: pipe, + } +} + // ============================================================================ // Script Integration (RTL Registration) // ============================================================================ @@ -439,7 +467,7 @@ pub fn register(env: &Environment) { "create-random-ohlc", StaticType::Function(Box::new(Signature { params: StaticType::Tuple(vec![StaticType::Int, StaticType::Int]), - ret: StaticType::Series(Box::new(StaticType::Record(ohlc_layout.clone()))), + ret: StaticType::Stream(Box::new(StaticType::Record(ohlc_layout.clone()))), })), Purity::Impure, // Modifies global generator registry move |args: &[Value]| { diff --git a/src/ast/types.rs b/src/ast/types.rs index 977dd14..540e741 100644 --- a/src/ast/types.rs +++ b/src/ast/types.rs @@ -311,6 +311,7 @@ pub enum StaticType { Optional(Box), // Represents T | Void (e.g. for filter pipes) List(Box), // Legacy / Dynamic list Series(Box), // Time series of a specific type + Stream(Box), // Reactive stream of a specific type Tuple(Vec), // Heterogeneous fixed-size Vector(Box, usize), // Homogeneous fixed-size Matrix(Box, Vec), // Multi-dimensional homogeneous @@ -337,6 +338,7 @@ impl fmt::Display for StaticType { StaticType::Optional(inner) => write!(f, "optional({})", inner), StaticType::List(inner) => write!(f, "list({})", inner), StaticType::Series(inner) => write!(f, "series<{}>", inner), + StaticType::Stream(inner) => write!(f, "stream<{}>", inner), StaticType::Tuple(elements) => { write!(f, "[")?; for (i, el) in elements.iter().enumerate() { @@ -440,6 +442,10 @@ impl StaticType { (StaticType::Series(inner_a), StaticType::Series(inner_b)) => { inner_a.is_assignable_from(inner_b) } + // Streams are assignable if their inner types are assignable + (StaticType::Stream(inner_a), StaticType::Stream(inner_b)) => { + inner_a.is_assignable_from(inner_b) + } _ => false, } } @@ -463,6 +469,20 @@ impl StaticType { return Some(layout.fields[idx].1.clone()); } } + StaticType::Series(inner) => { + if let StaticType::Record(layout) = inner.as_ref() + && let Some(idx) = layout.index_of(*k) + { + return Some(StaticType::Series(Box::new(layout.fields[idx].1.clone()))); + } + } + StaticType::Stream(inner) => { + if let StaticType::Record(layout) = inner.as_ref() + && let Some(idx) = layout.index_of(*k) + { + return Some(StaticType::Stream(Box::new(layout.fields[idx].1.clone()))); + } + } StaticType::Any => return Some(StaticType::Any), _ => {} } diff --git a/src/ast/vm.rs b/src/ast/vm.rs index 70e556c..236cdba 100644 --- a/src/ast/vm.rs +++ b/src/ast/vm.rs @@ -384,6 +384,12 @@ impl VM { field.name() )); } + } else if let Some(sn) = obj.as_any().downcast_ref::() { + let mapped = crate::ast::rtl::streams::build_map_stream(sn.inner.clone(), *field); + return Ok(Value::Object(Rc::new(mapped))); + } else if let Some(pn) = obj.as_any().downcast_ref::() { + let mapped = crate::ast::rtl::streams::build_map_stream(pn.stream.clone(), *field); + return Ok(Value::Object(Rc::new(mapped))); } // Fallback if it's another type of object that is not a RecordSeries. @@ -550,26 +556,33 @@ impl VM { k, ); return Ok(Value::Object(std::rc::Rc::new(view))); - } else { + } else { return Err(format!( "RecordSeries does not have field :{}", k.name() )); - } - } - return Err(format!( - "Field accessor .{} expects a record or RecordSeries, got {}", - k.name(), - obj.type_name() - )); - } else { - return Err(format!( - "Field accessor .{} expects a record or RecordSeries, got {}", - k.name(), - rec - )); - } - } + } + } else if let Some(sn) = obj.as_any().downcast_ref::() { + let mapped = crate::ast::rtl::streams::build_map_stream(sn.inner.clone(), k); + return Ok(Value::Object(Rc::new(mapped))); + } else if let Some(pn) = obj.as_any().downcast_ref::() { + let mapped = crate::ast::rtl::streams::build_map_stream(pn.stream.clone(), k); + return Ok(Value::Object(Rc::new(mapped))); + } + return Err(format!( + "Field accessor .{} expects a record, RecordSeries or Stream, got {}", + k.name(), + obj.type_name() + )); + } else { + return Err(format!( + "Field accessor .{} expects a record, RecordSeries or Stream, got {}", + k.name(), + rec + )); + } + } + _ => { return Err(format!( "Tail call target is not a function: {}", @@ -621,16 +634,22 @@ impl VM { k.name() )) } + } else if let Some(sn) = obj.as_any().downcast_ref::() { + let mapped = crate::ast::rtl::streams::build_map_stream(sn.inner.clone(), *k); + Ok(Value::Object(Rc::new(mapped))) + } else if let Some(pn) = obj.as_any().downcast_ref::() { + let mapped = crate::ast::rtl::streams::build_map_stream(pn.stream.clone(), *k); + Ok(Value::Object(Rc::new(mapped))) } else { Err(format!( - "Field accessor .{} expects a record or RecordSeries, got {}", + "Field accessor .{} expects a record, RecordSeries or Stream, got {}", k.name(), obj.type_name() )) } } else { Err(format!( - "Field accessor .{} expects a record or RecordSeries, got {}", + "Field accessor .{} expects a record, RecordSeries or Stream, got {}", k.name(), rec ))