diff --git a/examples/data_stream_print.myc b/examples/data_stream_print.myc
new file mode 100644
index 0000000..99b14d6
--- /dev/null
+++ b/examples/data_stream_print.myc
@@ -0,0 +1,8 @@
+;; Skip: data output to stdout
+(do
+  (def ohlc (create-m1-stream "EURUSD"))
+
+  ;; Sink: print returns void, so nothing is propagated downstream
+  (pipe ohlc
+    (fn [bar]
+      (print (.close bar)))))
diff --git a/src/ast/data_server/cache.rs b/src/ast/data_server/cache.rs
index e35aaf9..6bcd112 100644
--- a/src/ast/data_server/cache.rs
+++ b/src/ast/data_server/cache.rs
@@ -9,6 +9,9 @@ use super::records::{DataFormat, M1Parsed, TickParsed};
 use std::collections::{HashMap, HashSet};
 use std::sync::{Arc, Condvar, Mutex, RwLock};
 
+/// Per-symbol reference count for cache eviction.
+type SymbolRefCounts = Mutex<HashMap<Arc<str>, usize>>;
+
 /// Uniquely identifies one data file on disk.
 #[derive(Clone, Debug, Hash, Eq, PartialEq)]
 pub struct FileKey {
@@ -40,6 +43,9 @@ pub struct FileCache {
     /// Signaled whenever a file finishes loading (insert completes).
     /// Waiters re-check `is_cached` after each wake-up.
     loaded: Condvar,
+    /// Per-symbol reference counts. When a count drops to zero, all
+    /// cached files for that symbol are evicted from both maps.
+    symbol_refs: SymbolRefCounts,
 }
 
 impl Default for FileCache {
@@ -55,6 +61,7 @@ impl FileCache {
             tick: RwLock::new(HashMap::new()),
             loading: Mutex::new(HashSet::new()),
             loaded: Condvar::new(),
+            symbol_refs: Mutex::new(HashMap::new()),
         }
     }
 
@@ -132,6 +139,46 @@ impl FileCache {
             guard = self.loaded.wait(guard).unwrap();
         }
     }
+
+    // -- Symbol-level reference counting ------------------------------------
+
+    /// Increments the reference count for a symbol. Call this when creating
+    /// a new [`SymbolChunkIter`] so the cached data stays alive.
+    pub fn retain_symbol(&self, symbol: &Arc<str>) {
+        *self
+            .symbol_refs
+            .lock()
+            .unwrap()
+            .entry(Arc::clone(symbol))
+            .or_insert(0) += 1;
+    }
+
+    /// Decrements the reference count for a symbol. When the count reaches
+    /// zero, all cached files (M1 and tick) for that symbol are evicted.
+    pub fn release_symbol(&self, symbol: &Arc<str>) {
+        let mut refs = self.symbol_refs.lock().unwrap();
+        if let Some(count) = refs.get_mut(symbol) {
+            *count -= 1;
+            if *count == 0 {
+                refs.remove(symbol);
+                // Drop the lock before acquiring write locks on the caches
+                drop(refs);
+                self.evict_symbol(symbol);
+            }
+        }
+    }
+
+    /// Removes all cached entries for the given symbol from both maps.
+    fn evict_symbol(&self, symbol: &Arc<str>) {
+        self.m1
+            .write()
+            .unwrap()
+            .retain(|key, _| key.symbol != *symbol);
+        self.tick
+            .write()
+            .unwrap()
+            .retain(|key, _| key.symbol != *symbol);
+    }
 }
 
 #[cfg(test)]
@@ -233,4 +280,39 @@ mod tests {
         // Waiter should now see the cached data
         assert!(handle.join().unwrap());
     }
+
+    #[test]
+    fn test_symbol_retain_release() {
+        let cache = FileCache::new();
+        let sym: Arc<str> = Arc::from("EURUSD");
+        let key1 = test_key("EURUSD", 2017, 1, DataFormat::M1);
+        let key2 = test_key("EURUSD", 2017, 2, DataFormat::M1);
+        let key3 = test_key("EURUSD", 2017, 1, DataFormat::Tick);
+        let other = test_key("GBPUSD", 2020, 6, DataFormat::M1);
+
+        // Insert data for both symbols
+        cache.insert_m1(key1.clone(), vec![]);
+        cache.insert_m1(key2.clone(), vec![]);
+        cache.insert_tick(key3.clone(), vec![]);
+        cache.insert_m1(other.clone(), vec![]);
+
+        // Two consumers retain EURUSD
+        cache.retain_symbol(&sym);
+        cache.retain_symbol(&sym);
+
+        // First release: refcount drops to 1 — data stays
+        cache.release_symbol(&sym);
+        assert!(cache.is_cached(&key1));
+        assert!(cache.is_cached(&key2));
+        assert!(cache.is_cached(&key3));
+
+        // Second release: refcount drops to 0 — all EURUSD evicted
+        cache.release_symbol(&sym);
+        assert!(!cache.is_cached(&key1));
+        assert!(!cache.is_cached(&key2));
+        assert!(!cache.is_cached(&key3));
+
+        // GBPUSD is untouched
+        assert!(cache.is_cached(&other));
+    }
 }
diff --git a/src/ast/data_server/mod.rs b/src/ast/data_server/mod.rs
index 6cd4e46..eae5083 100644
--- a/src/ast/data_server/mod.rs
+++ b/src/ast/data_server/mod.rs
@@ -190,6 +190,7 @@ impl DataServer {
     /// The iterator yields `Arc<[M1Parsed]>` chunks in chronological order,
     /// prefetching the next file while the current one is consumed.
     pub fn stream_m1(self: &Arc<Self>, symbol: &str) -> Option<SymbolChunkIter<M1Parsed>> {
+        let symbol_arc = self.index.symbols.get_key_value(symbol)?.0.clone();
         let files: Vec<FileKey> = self
             .index
             .symbols
@@ -209,8 +210,11 @@ impl DataServer {
             self.prefetch_m1(&files[1]);
         }
 
+        self.cache.retain_symbol(&symbol_arc);
+
         Some(SymbolChunkIter {
             server: Arc::clone(self),
+            symbol: symbol_arc,
             files,
             file_idx: 0,
             chunk_idx: 0,
@@ -220,6 +224,7 @@ impl DataServer {
 
     /// Returns a chunk iterator over all tick data for a symbol.
     pub fn stream_tick(self: &Arc<Self>, symbol: &str) -> Option<SymbolChunkIter<TickParsed>> {
+        let symbol_arc = self.index.symbols.get_key_value(symbol)?.0.clone();
         let files: Vec<FileKey> = self
             .index
             .symbols
@@ -238,8 +243,11 @@ impl DataServer {
             self.prefetch_tick(&files[1]);
         }
 
+        self.cache.retain_symbol(&symbol_arc);
+
         Some(SymbolChunkIter {
             server: Arc::clone(self),
+            symbol: symbol_arc,
             files,
             file_idx: 0,
             chunk_idx: 0,
@@ -346,6 +354,8 @@ impl DataServer {
 /// `LoadDataFile(nextFileInfo)` prefetch call.
 pub struct SymbolChunkIter<T> {
     server: Arc<DataServer>,
+    /// Symbol name, held for `Drop`-based cache release.
+    symbol: Arc<str>,
     files: Vec<FileKey>,
     file_idx: usize,
     chunk_idx: usize,
@@ -354,6 +364,12 @@ pub struct SymbolChunkIter<T> {
     current_file_chunks: Option<ChunkVec<T>>,
 }
 
+impl<T> Drop for SymbolChunkIter<T> {
+    fn drop(&mut self) {
+        self.server.cache.release_symbol(&self.symbol);
+    }
+}
+
 impl SymbolChunkIter<M1Parsed> {
     /// Returns the next chunk of M1 records, or `None` when exhausted.
     pub fn next_chunk(&mut self) -> Option<Arc<[M1Parsed]>> {