security: anchor HTF aggregation on the symbol clock (syminfo.timezone + session)

include/pineforge/timeframe.hpp

@@ -85,6 +85,18 @@ CalendarPeriod calendar_period_for(const std::string& tf);
 /// Check if two timestamps (Unix milliseconds) fall in different calendar periods.
 bool crosses_boundary(int64_t prev_ms, int64_t curr_ms, CalendarPeriod period);
 
+/// Timezone/session-aware variants. The tz-less forms above evaluate the
+/// calendar in UTC and intraday buckets on the epoch grid — exactly TV's
+/// behavior for 24x7 symbols (the corpus regime). Session symbols
+/// (equities RTH, forex) anchor HTF buckets on the exchange clock instead:
+/// daily boundaries at symbol-local midnight, intraday buckets offset by the
+/// session-open minutes. With tz="UTC" and session ""/"24x7" these are
+/// bit-identical to the UTC forms, so existing callers are unaffected.
+bool crosses_boundary(int64_t prev_ms, int64_t curr_ms, CalendarPeriod period,
+                      const std::string& tz, const std::string& session);
+bool tf_change(int64_t prev_ms, int64_t curr_ms, const std::string& tf,
+               const std::string& tz, const std::string& session);
+
 // ─── TimeframeAggregator ───────────────────────────────────────────────────────
 
 class TimeframeAggregator {
@@ -99,6 +111,15 @@ class TimeframeAggregator {
     TimeframeAggregator(const std::string& target_tf,
                         const std::string& input_tf);
 
+    /// Calendar/ratio aggregation anchored on a symbol clock. tz is the Pine
+    /// syminfo.timezone (exchange tz); session the Pine session string
+    /// ("0930-1600", "24x7", ...). Defaults reproduce the UTC/24x7 forms
+    /// bit-for-bit, so every existing construction site compiles unchanged.
+    TimeframeAggregator(const std::string& target_tf,
+                        const std::string& input_tf,
+                        const std::string& tz,
+                        const std::string& session = "");
+
     /// Feed one input bar. Returns aggregation state.
     AggregatedBar feed(const Bar& input_bar);
 
@@ -119,6 +140,8 @@ class TimeframeAggregator {
     CalendarPeriod cal_period_ = CalendarPeriod::NONE; // for CALENDAR mode
     int64_t target_seconds_ = 0;       // wall-clock seconds for RATIO boundary detection
     int64_t input_seconds_ = 0;        // input bar duration (seconds), when known
+    std::string anchor_tz_ = "UTC";    // syminfo.timezone (exchange clock)
+    std::string anchor_session_;       // syminfo.session ("" or "24x7" = none)
 
     Bar current_bar_{};
     Bar last_completed_bar_{};

src/engine_run.cpp

@@ -1196,7 +1196,8 @@ void BacktestEngine::run(const Bar* input_bars, int n_input,
     if (needs_aggregation) {
         // Use a single timeframe-constructor path so script timeframe boundaries
         // follow the same wall-clock/calendar semantics as request.security.
-        script_tf_agg_ = TimeframeAggregator(effective_script_tf, effective_input_tf);
+        script_tf_agg_ = TimeframeAggregator(effective_script_tf, effective_input_tf,
+                                             syminfo_.timezone, syminfo_.session);
     } else {
         script_tf_agg_ = TimeframeAggregator();  // passthrough
     }
@@ -1274,9 +1275,11 @@ void BacktestEngine::init_security_eval_states_for_run(
         }
         int req_ratio = tf_ratio(effective_input_tf, state.tf);
         if (req_ratio > 1) {
-            state.aggregator = TimeframeAggregator(state.tf, effective_input_tf);
+            state.aggregator = TimeframeAggregator(state.tf, effective_input_tf,
+                syminfo_.timezone, syminfo_.session);
         } else if (req_ratio == -1) {
-            state.aggregator = TimeframeAggregator(state.tf, effective_input_tf);
+            state.aggregator = TimeframeAggregator(state.tf, effective_input_tf,
+                syminfo_.timezone, syminfo_.session);
         }
     }
 }
@@ -1358,7 +1361,8 @@ void BacktestEngine::prepare_historical_security_lookahead_projections(
             // tests/synthetic feeds beginning at zero correct via the fixed-TF
             // bucket fallback; real feeds take the calendar-aware path.
             if (from_ms != 0 && to_ms != 0) {
-                return tf_change(from_ms, to_ms, state.tf);
+                return tf_change(from_ms, to_ms, state.tf,
+                                 syminfo_.timezone, syminfo_.session);
             }
             const int64_t requested_ms =
                 static_cast<int64_t>(requested_seconds) * 1000;

src/engine_security.cpp

@@ -4,7 +4,9 @@
 
 #include "engine_internal.hpp"
 
+#include <pineforge/session_time.hpp>
 #include <pineforge/ta.hpp>
+#include <pineforge/timeframe.hpp>
 
 #include <algorithm>
 #include <cctype>
@@ -13,6 +15,30 @@
 #include <unordered_set>
 
 namespace pineforge {
+namespace {
+// Mirror of the aggregator's intraday clock in timeframe.cpp: ms that the
+// symbol clock (tz + session open) has removed from the epoch grid at `ts`.
+// Must stay arithmetic-identical to TimeframeAggregator's intraday_clock_ms().
+int64_t syminfo_clock_shift_ms(int64_t ts, const std::string& tz,
+                               const std::string& session) {
+    if ((tz.empty() || tz == "UTC" || tz == "Etc/UTC")
+        && (session.empty() || session == "24x7")) return 0;
+    int64_t day_open = calendar_day_open_local_ms(ts, tz);
+    int64_t shift = day_open - (day_open / kMsPerDay) * kMsPerDay;
+    if (!(session.empty() || session == "24x7")) {
+        int digits = 0, value = 0;
+        for (char c : session) {
+            if (c >= '0' && c <= '9') {
+                value = value * 10 + (c - '0');
+                if (++digits >= 4) break;
+            } else if (digits > 0) break;
+        }
+        shift -= static_cast<int64_t>((value / 100) * 60 + (value % 100)) * 60000;
+    }
+    return shift;
+}
+}  // namespace
+
 using namespace internal;
 
 
@@ -40,9 +66,11 @@ void BacktestEngine::register_security_eval(int sec_id, const std::string& reque
         } else {
             int ratio = tf_ratio(input_tf, requested_tf);
             if (ratio > 1) {
-                state.aggregator = TimeframeAggregator(requested_tf, input_tf);
+                state.aggregator = TimeframeAggregator(requested_tf, input_tf,
+                    syminfo_.timezone, syminfo_.session);
             } else if (ratio == -1) {
-                state.aggregator = TimeframeAggregator(requested_tf, input_tf);
+                state.aggregator = TimeframeAggregator(requested_tf, input_tf,
+                    syminfo_.timezone, syminfo_.session);
             }
             // ratio <= 0: passthrough (same or unsupported lower TF)
         }
@@ -512,8 +540,14 @@ void BacktestEngine::feed_security_eval_state(SecurityEvalState& state, const Ba
         // bookkeeping above stays driven by the real completion.
         bool publish = true;
         if (state.publish_gate_tf_seconds > 0 && script_tf_seconds_ > 0) {
+            // Anchor on the symbol clock (the same clock as the aggregators):
+            // exchange-tz seconds since local-midnight+session-open. With
+            // tz=UTC and no session the shift is zero, reducing to the
+            // previous epoch math bit-for-bit.
+            int64_t shifted = ab.bar.timestamp - syminfo_clock_shift_ms(
+                ab.bar.timestamp, syminfo_.timezone, syminfo_.session);
             int64_t bucket_end_sec =
-                ab.bar.timestamp / 1000 + state.publish_gate_tf_seconds;
+                shifted / 1000 + state.publish_gate_tf_seconds;
             publish = (bucket_end_sec % script_tf_seconds_) == 0;
         }
         evaluate_security(state.sec_id, ab.bar, publish);

src/engine_stream.cpp

@@ -299,7 +299,8 @@ void BacktestEngine::stream_feed_input_bar(const Bar& bar, bool had_tick) {
 
     AggregatedBar ab = script_tf_agg_.feed(bar);
     const bool completed_on_boundary = ab.is_complete
-        && tf_change(ab.bar.timestamp, bar.timestamp, script_tf_);
+        && tf_change(ab.bar.timestamp, bar.timestamp, script_tf_,
+                     syminfo_.timezone, syminfo_.session);
     if (completed_on_boundary) {
         // The current input bar opened the next bucket; the aggregator emitted
         // the preceding partial bucket before retaining this bar as its new