pine_scheduler.hpp

Go to the documentation of this file.

#pragma once
 
#include <pineforge/native_host.hpp>
#include <pineforge/source/pine_language_state.hpp>
 
#include <algorithm>
#include <array>
#include <cstdint>
#include <limits>
#include <memory>
#include <optional>
#include <string>
#include <vector>
 
namespace pineforge {
inline namespace engine_script_run_v19 { class ScriptBucketCursor; }
}
 
namespace pineforge::source {
 
class PineStrategyHost;
 
// Source cadence over generic native callbacks. The driver remains the sole
// owner of matching; this class owns only language publication, retained begin
// data and the Pine calc-on-order-fills callback cadence.
class PineScheduler {
public:
    PineScheduler();
    ~PineScheduler();
 
    // C++17 read-only span. Empty iterators use one sentinel so standard
    // pointer-range algorithms never subtract two null pointers; data()
    // still exposes the released span as exactly {nullptr, 0}.
    class InputView {
    public:
        InputView() noexcept = default;
        InputView(const Bar* bars, std::size_t n) noexcept : data_(bars), size_(n) {}
        const Bar* data() const noexcept { return data_; }
        std::size_t size() const noexcept { return size_; }
        bool empty() const noexcept { return size_ == 0; }
        const Bar& operator[](std::size_t i) const noexcept { return data_[i]; }
        const Bar& front() const noexcept { return data_[0]; }
        const Bar& back() const noexcept { return data_[size_ - 1]; }
        const Bar* begin() const noexcept { return data_ ? data_ : &empty_bar_; }
        const Bar* end() const noexcept { return begin() + size_; }
    private:
        const Bar* data_ = nullptr;
        std::size_t size_ = 0;
        inline static constexpr Bar empty_bar_{};
    };
 
    void capture_begin(const NativeBeginArgs&);
    // Fold the consumed prefix while caller memory is alive, then detach it.
    // A stream keeps its owned warmup until the next capture/destruction.
    void release_batch_input() noexcept;
    void run_begin(PineStrategyHost&);
    void input(const Bar&, const NativeInputContext&, PineStrategyHost&);
    void tick(const Bar&, const NativeTickContext&, PineStrategyHost&);
    void bar_open(const Bar&, const NativeDecisionContext&, PineStrategyHost&);
    void bar(const Bar&, const NativeDecisionContext&, PineStrategyHost&);
    // Applied-notification bookkeeping only. The calc_on_order_fills
    // recalculation itself is the kernel's cadence now
    // (NativeCalculationTrigger::BarCloseAndFills): the consumer drives it
    // from the same drain iteration, right after this callback returns, and
    // delivers it as recalculate() below.
    void applied(const native_order::ExecutionAppliedEvent&);
    // Pine's own admission of a fill recalculation, re-checked at the
    // kernel's OrderFill cursor. Pure: it reads the freshly advanced applied
    // cursor, the source config and adapter facts, and mutates nothing.
    bool coof_recalculation_due(const native_order::ExecutionAppliedEvent&,
                                const NativeDecisionContext&, PineStrategyHost&) const;
    void recalculate(const native_order::ExecutionAppliedEvent&, const NativeDecisionContext&,
                     PineStrategyHost&);
 
    PineLanguageState& language() noexcept { return language_; }
    bool is_first_tick() const noexcept { return language_.is_first_tick_; }
    bool is_last_tick() const noexcept { return language_.is_last_tick_; }
    bool bar_magnifier_enabled() const noexcept { return retained_.bar_magnifier; }
    bool history_advances_new_bar() const noexcept {
        return language_.is_first_tick_ && language_.history_slot_is_new_;
    }
    bool security_series_slot_is_new(int) const noexcept {
        return language_.history_slot_is_new_;
    }
    double previous_chart_close() const noexcept { return language_.prev_chart_close_; }
    int bar_index_offset() const noexcept { return language_.bar_index_offset_; }
    void set_bar_index_offset(int value) noexcept { language_.bar_index_offset_ = value; }
    void set_source_series_active(bool value) noexcept { language_._src_series_active_ = value; }
    double script_position_view(int bar_index, PositionSide side, double quantity) const noexcept;
    void freeze_script_position_view(int bar_index, PositionSide side, double quantity,
                                     const std::vector<PyramidEntry>& lots);
    void clear_script_position_view() noexcept;
    const Series<double>& source_series(const std::string&) const;
    void fixture_publish_source_series(const Bar&, bool new_history_slot);
    int source_bar_count() const noexcept { return source_bar_count_; }
    std::optional<Bar> next_source_bar(int interval_index) const {
        if (retained_.is_stream || retained_.bar_magnifier
            || (!retained_.input_tf.empty() && !retained_.script_tf.empty()
                && retained_.input_tf != retained_.script_tf)
            || interval_index < 0
            || interval_index + 1 >= static_cast<int>(retained_.bars.size())) {
            return std::nullopt;
        }
        return retained_.bars[static_cast<std::size_t>(interval_index + 1)];
    }
    // The first input the run retains -- a batch's, or a stream's warmup --
    // after input `input_index` that opens at or after `at_or_after`, or
    // nullopt when the retained input ends first.
    std::optional<Bar> retained_input_from(int input_index, std::int64_t at_or_after) const {
        if (input_index < 0) return std::nullopt;
        for (std::size_t i = static_cast<std::size_t>(input_index) + 1; i < retained_.bars.size(); ++i) {
            if (retained_.bars[i].timestamp >= at_or_after) return retained_.bars[i];
        }
        return std::nullopt;
    }
    bool retains_stream() const noexcept { return retained_.is_stream; }
    bool input_is_observed_ticks() const noexcept { return input_is_observed_ticks_; }
    // broker_bar is used only by observe_intraday_cap_fill/noop: their context
    // names the current sub-bar, or the prior one while its close is dispatched.
    // Thus two recent inputs suffice; eight retain conservative headroom, not
    // arbitrary history. Two separate opening anchors cover the current and
    // preceding script intervals even when either contains more than eight inputs.
    void retain_confirmed_input(const Bar& bar, std::int64_t script_open_ms =
            std::numeric_limits<std::int64_t>::min()) noexcept {
        if (!retained_.is_stream) return;
        if (bar.timestamp == script_open_ms) {
            bool replaced = false;
            for (std::size_t index = 0; index < confirmed_open_input_count_; ++index) {
                if (confirmed_open_input_bars_[index].timestamp == bar.timestamp) {
                    confirmed_open_input_bars_[index] = bar;
                    replaced = true;
                    break;
                }
            }
            if (!replaced) {
                confirmed_open_input_bars_[confirmed_open_input_next_] = bar;
                confirmed_open_input_next_ =
                    (confirmed_open_input_next_ + 1) % confirmed_open_input_bars_.size();
                confirmed_open_input_count_ =
                    std::min(confirmed_open_input_count_ + 1, confirmed_open_input_bars_.size());
            }
        }
        for (std::size_t index = 0; index < confirmed_input_count_; ++index) {
            auto& existing = confirmed_input_bars_[index];
            if (existing.timestamp == bar.timestamp) {
                existing = bar;
                return;
            }
        }
        confirmed_input_bars_[confirmed_input_next_] = bar;
        confirmed_input_next_ = (confirmed_input_next_ + 1) % confirmed_input_bars_.size();
        confirmed_input_count_ = std::min(confirmed_input_count_ + 1, confirmed_input_bars_.size());
    }
    std::size_t confirmed_input_count() const noexcept { return confirmed_input_count_; }
    std::optional<Bar> confirmed_input_bar_at(std::int64_t timestamp) const noexcept {
        for (std::size_t index = 0; index < confirmed_input_count_; ++index) {
            if (confirmed_input_bars_[index].timestamp == timestamp)
                return confirmed_input_bars_[index];
        }
        for (std::size_t index = 0; index < confirmed_open_input_count_; ++index) {
            if (confirmed_open_input_bars_[index].timestamp == timestamp)
                return confirmed_open_input_bars_[index];
        }
        return std::nullopt;
    }
    static bool confirmed_script_interval_complete(const NativeInputContext& context) noexcept {
        return context.script_interval.last_traded_close_ms > context.script_interval.open_ms
            && context.input_interval.last_traded_close_ms
                >= context.script_interval.last_traded_close_ms;
    }
    // The input the run retains -- a batch's, or a stream's warmup -- in
    // time order, and whether each of its bars is a chart bar (no
    // aggregation to a coarser script timeframe).
    InputView retained_input() const noexcept { return retained_.bars; }
    bool retained_input_is_chart() const noexcept {
        return retained_.input_tf.empty() || retained_.script_tf.empty()
            || retained_.input_tf == retained_.script_tf;
    }
    // The open of the script bar published last, before the run's first none.
    std::optional<std::int64_t> last_published_script_open() const noexcept {
        if (last_published_script_open_ms_ == std::numeric_limits<std::int64_t>::min())
            return std::nullopt;
        return last_published_script_open_ms_;
    }
    bool terminal_source_bar() const noexcept {
        return expected_source_bars_ > 0 && source_bar_count_ >= expected_source_bars_;
    }
    int source_bar_index_for(const NativeDecisionContext& context) const noexcept;
    // The next fill point, for a fill recalculation's market order, of the
    // lower-path bar being walked: a price that bar still reaches, or
    // (next_open) the next lower-path bar's opening -- NaN when there is
    // none -- which a plain market order reaches by itself, since the kernel
    // fills a newborn market request at the path's next discrete point. The
    // leg order is the kernel's answer for that bar
    // (PineExecutionAdapter::source_path_uses_high_first), not a copy of it.
    struct InputWaypoint {
        double price;
        bool next_open;
    };
    std::optional<InputWaypoint> next_input_waypoint(
        const PineStrategyHost&, const NativeDecisionContext&) const noexcept;
    const Bar* current_script_bar() const noexcept {
        return current_script_bar_valid_ ? &current_script_bar_ : nullptr;
    }
    // The bar of the run's lower intrabar path the context is walking (its
    // sub_bar_open_ms), or nullptr when the run declares no lower path or no
    // bar of it carries that stamp. A magnified run's sub-bars are
    // TradingView's intrabars built from the input (R5 lane MAG-INTRABAR,
    // source::tradingview_magnifier_bars), not input bars.
    const Bar* lower_path_bar_at(const PineStrategyHost& host,
                                 const NativeDecisionContext& context) const noexcept;
    // One intrabar of a magnified run's path is done. TradingView's magnified
    // broker admits the open fill and one refill at EVERY intrabar's open,
    // not only at the chart bar's first one (R5 lane MAG-INTRABAR,
    // tests/fixtures/magnifier_intrabars/mi-coof-refill*), so the open-point
    // count starts over with the next intrabar.
    void sub_bar_complete() noexcept;
    std::optional<Bar> broker_bar(const PineStrategyHost& host,
                                  const NativeDecisionContext& context) const {
        if (const Bar* sub = lower_path_bar_at(host, context)) return *sub;
        if (const auto observed = confirmed_input_bar_at(context.sub_bar_open_ms)) return observed;
        const auto& bars = retained_.bars;
        const std::size_t n = bars.size();
        if (n > 0) {
            if (broker_bar_cursor < n && bars[broker_bar_cursor].timestamp == context.sub_bar_open_ms) {
                return bars[broker_bar_cursor];
            }
            if (broker_bar_cursor >= n || bars[broker_bar_cursor].timestamp > context.sub_bar_open_ms) {
                broker_bar_cursor = 0;
            }
            while (broker_bar_cursor < n && bars[broker_bar_cursor].timestamp < context.sub_bar_open_ms) {
                ++broker_bar_cursor;
            }
            if (broker_bar_cursor < n && bars[broker_bar_cursor].timestamp == context.sub_bar_open_ms) {
                return bars[broker_bar_cursor];
            }
            const auto found = std::find_if(bars.begin(), bars.end(),
                [&](const Bar& bar) { return bar.timestamp == context.sub_bar_open_ms; });
            if (found != bars.end()) {
                broker_bar_cursor = static_cast<std::size_t>(std::distance(bars.begin(), found));
                return *found;
            }
        }
        if (retained_.is_stream && !input_is_observed_ticks_) return std::nullopt;
        return current_script_bar_valid_ ? std::optional<Bar>{current_script_bar_}
                                         : std::nullopt;
    }
 
    void hash_state(BrokerStateHashSink&) const;
 
private:
    struct RetainedBegin {
        InputView bars;
        std::vector<Bar> owned;
        std::string input_tf;
        std::string script_tf;
        bool bar_magnifier = false;
        int magnifier_samples = 4;
        MagnifierDistribution distribution = MagnifierDistribution::ENDPOINTS;
        bool volume_weighted = false;
        int volume_weighted_min_samples = 2;
        int volume_weighted_max_samples = 64;
        bool is_stream = false;
        int warmup_n = 0;
        bool simple_run = false;
    };
 
    void publish_series(const Bar&, PineStrategyHost&);
    void update_source_series(const Bar&);
    void reset_language();
    // Restarts the consumed-prefix digests below (the retained input changed).
    void reset_consumed_digests() noexcept;
    void fold_consumed_digests() const noexcept;
    void snapshot_coof_script_state(PineStrategyHost&);
    void restore_coof_script_state(PineStrategyHost&);
    void commit_coof_script_state(PineStrategyHost&);
 
    struct DeferredBoundaryInput {
        Bar bar{};
        std::int64_t next_input_ms = 0;
        std::int64_t prior_script_open_ms = 0;
        bool calling_bar_complete = false;
        bool all_security_states = false;
        bool active = false;
    };
 
    // Derived lookup cursor over retained_.bars (reset-and-rescan in
    // broker_bar()); waived from the state hash by
    // scripts/check_broker_state_hash_coverage.py::SCHEDULER_CACHE_WAIVERS.
    mutable std::size_t broker_bar_cursor = 0;
 
    // Derived completion lookup over retained_.bars. Repeated or backward
    // input indices are answered by the cursor's last result or a rescan.
    // Waived as a cache; the independent consumed-prefix cursor folds below.
    mutable std::unique_ptr<ScriptBucketCursor> input_script_cursor;
 
    // @source-state begin
    PineLanguageState language_;
    RetainedBegin retained_;
    // The hash retains its old consumed count after bars becomes {nullptr,0}.
    // Unconsumed future input remains transport, as in the v3 hash domain.
    std::size_t retained_n_ = 0;
    std::int64_t current_script_open_ms_ = 0;
    Bar current_script_bar_{};
    bool current_script_bar_valid_ = false;
    bool saw_open_fill_ = false;
    int open_point_fills_ = 0;
    int source_bar_count_ = 0;
    int expected_source_bars_ = 0;
    std::uint64_t applied_cursor_ = 0;
    std::int64_t coof_callback_script_open_ = std::numeric_limits<std::int64_t>::min();
    std::int64_t last_published_script_open_ms_ =
        std::numeric_limits<std::int64_t>::min();
    std::int64_t prior_input_script_open_ms_ = std::numeric_limits<std::int64_t>::min();
    std::int64_t awaiting_legacy_script_open_ms_ = std::numeric_limits<std::int64_t>::min();
    std::int64_t last_stream_input_open_ms_ = std::numeric_limits<std::int64_t>::min();
    // The run-begin preview's extent, including unconsumed transport. Only
    // its consumed prefix is hashed, just as for the former flag vectors.
    std::size_t input_script_size_ = 0;
    bool uses_aux_security_feed_ = false;
    bool input_is_observed_ticks_ = false;
    std::array<Bar, 8> confirmed_input_bars_{};
    std::size_t confirmed_input_next_ = 0;
    std::size_t confirmed_input_count_ = 0;
    std::array<Bar, 2> confirmed_open_input_bars_{};
    std::size_t confirmed_open_input_next_ = 0;
    std::size_t confirmed_open_input_count_ = 0;
    DeferredBoundaryInput deferred_boundary_input_{};
    // R5 lane V19-E: running digests of the consumed input prefix -- the bars,
    // their script completions and their boundary completions -- each element
    // folded once, the first time the state hash reads past it (the v2
    // scheduler fold re-read the whole prefix at every read).
    mutable std::size_t consumed_bars_folded_ = 0;
    mutable std::uint64_t consumed_bars_digest_ = 1469598103934665603ULL;
    mutable std::size_t consumed_completes_folded_ = 0;
    mutable std::uint64_t consumed_completes_digest_ = 1469598103934665603ULL;
    mutable std::size_t consumed_boundaries_folded_ = 0;
    mutable std::uint64_t consumed_boundaries_digest_ = 1469598103934665603ULL;
    mutable std::unique_ptr<ScriptBucketCursor> consumed_script_cursor_;
    // @source-state end
};
 
} // namespace pineforge::source