|
| 1 | +#pragma once |
| 2 | + |
| 3 | +#include "pine_position_tail.hpp" |
| 4 | + |
| 5 | +#include <pineforge/detail/adapter_exposure_storage.hpp> |
| 6 | +#include <pineforge/detail/disk_id_table.hpp> |
| 7 | +#include <pineforge/engine.hpp> |
| 8 | +#include <pineforge/source/pine_adapter.hpp> |
| 9 | + |
| 10 | +#include <algorithm> |
| 11 | +#include <cmath> |
| 12 | +#include <cstddef> |
| 13 | +#include <cstdint> |
| 14 | +#include <limits> |
| 15 | +#include <string> |
| 16 | +#include <unordered_map> |
| 17 | +#include <unordered_set> |
| 18 | + |
| 19 | +namespace pineforge::source::detail { |
| 20 | + |
| 21 | +// A proof, not a differently rounded floating-point accumulator. Each term |
| 22 | +// is a nonnegative integer times 2^quantum; the numerator of the WHOLE |
| 23 | +// selected sum must be strictly below 2^53. Every subset/partial sum is then |
| 24 | +// exactly representable, including the original newest-first left fold. |
| 25 | +// Reducing the quantum checks integer shifts BEFORE doing them. No rounded |
| 26 | +// subtotal, cancellation or subtraction of a retired origin is used. |
| 27 | +struct PositionTailExact { |
| 28 | + static constexpr std::uint64_t maximum = (UINT64_C(1) << 53) - 1; |
| 29 | + std::uint64_t magnitude = 0; |
| 30 | + std::int16_t quantum = 0; |
| 31 | + bool valid = true; |
| 32 | + |
| 33 | + void add(double value) noexcept { |
| 34 | + if (!valid) return; |
| 35 | + if (!std::isfinite(value) || std::signbit(value)) { |
| 36 | + valid = false; |
| 37 | + return; |
| 38 | + } |
| 39 | + if (value == 0.0) return; |
| 40 | + int exponent = 0; |
| 41 | + const double fraction = std::frexp(value, &exponent); |
| 42 | + auto integer = static_cast<std::uint64_t>(std::ldexp(fraction, 53)); |
| 43 | + exponent -= 53; |
| 44 | + while ((integer & 1) == 0) { |
| 45 | + integer >>= 1; |
| 46 | + ++exponent; |
| 47 | + } |
| 48 | + PositionTailExact term; |
| 49 | + term.magnitude = integer; |
| 50 | + term.quantum = static_cast<std::int16_t>(exponent); |
| 51 | + merge(term); |
| 52 | + } |
| 53 | + |
| 54 | + void merge(const PositionTailExact& other) noexcept { |
| 55 | + if (!valid || !other.valid) { |
| 56 | + valid = false; |
| 57 | + return; |
| 58 | + } |
| 59 | + if (other.magnitude == 0) return; |
| 60 | + if (magnitude == 0) { |
| 61 | + *this = other; |
| 62 | + return; |
| 63 | + } |
| 64 | + const int common = std::min(quantum, other.quantum); |
| 65 | + std::uint64_t left = magnitude, right = other.magnitude; |
| 66 | + const auto shift = [](std::uint64_t& value, int bits) { |
| 67 | + if (bits >= 53 || value > (maximum >> bits)) return false; |
| 68 | + value <<= bits; |
| 69 | + return true; |
| 70 | + }; |
| 71 | + if (!shift(left, quantum - common) || !shift(right, other.quantum - common) |
| 72 | + || right > maximum - left) { |
| 73 | + valid = false; |
| 74 | + return; |
| 75 | + } |
| 76 | + magnitude = left + right; |
| 77 | + quantum = static_cast<std::int16_t>(common); |
| 78 | + } |
| 79 | + |
| 80 | + bool read(double& result) const noexcept { |
| 81 | + if (!valid) return false; |
| 82 | + const double value = std::ldexp(static_cast<double>(magnitude), quantum); |
| 83 | + if (!std::isfinite(value)) return false; |
| 84 | + result = value; |
| 85 | + return true; |
| 86 | + } |
| 87 | +}; |
| 88 | +static_assert(sizeof(PositionTailExact) == 16, "compact dyadic certificate"); |
| 89 | +static_assert(std::numeric_limits<double>::is_iec559 |
| 90 | + && std::numeric_limits<double>::digits == 53, "binary64 proof"); |
| 91 | + |
| 92 | +class PositionTailSummary { |
| 93 | + struct Origin { |
| 94 | + PositionTailExact completed; |
| 95 | + std::uint32_t visited = 0; |
| 96 | + bool is_long = false; |
| 97 | + }; |
| 98 | + using Origins = std::unordered_map<std::uint64_t, Origin>; |
| 99 | + static_assert(sizeof(Origins::value_type) == 32, "32 bytes per origin, excluding map overhead"); |
| 100 | + struct Margins { |
| 101 | + PositionTailExact side[2]; |
| 102 | + }; |
| 103 | + |
| 104 | +public: |
| 105 | + PineExecutionAdapter::PositionTailWork work{}; |
| 106 | + |
| 107 | + void set_directory(const std::string& directory) { margins_.set_directory(directory); } |
| 108 | + |
| 109 | + // This cache can be discarded at any point: uncertified/missing facts |
| 110 | + // ALWAYS select replay. It owns no logical broker transition. Reset |
| 111 | + // rebuilds reachable equal-time rows from the current row source rather |
| 112 | + // than assuming an empty position tail at an opening/reversal. |
| 113 | + template<class Read> |
| 114 | + void begin_position(std::int64_t barrier, std::size_t base, std::size_t end, Read&& read) { |
| 115 | + forget_origins(); |
| 116 | + margins_.clear(); |
| 117 | + all_margins_ = {}; |
| 118 | + high_closed_origin_ = 0; |
| 119 | + barrier_ = barrier; |
| 120 | + observed_end_ = end; |
| 121 | + active_ = true; |
| 122 | + safe_ = true; |
| 123 | + have_time_ = false; |
| 124 | + std::int64_t newer_time = 0; |
| 125 | + for (std::size_t index = end; index > base;) { |
| 126 | + const Trade& row = read(--index); |
| 127 | + ++work.update_rows; |
| 128 | + if (row.exit_time < barrier) break; // strict barrier, equality survives |
| 129 | + if (have_time_ && row.exit_time > newer_time) { |
| 130 | + invalidate(); |
| 131 | + break; |
| 132 | + } |
| 133 | + if (!have_time_) last_time_ = row.exit_time; |
| 134 | + have_time_ = true; |
| 135 | + newer_time = row.exit_time; |
| 136 | + accumulate(row); |
| 137 | + } |
| 138 | + } |
| 139 | + |
| 140 | + template<class Read> |
| 141 | + void observe(std::size_t base, std::size_t end, Read&& read) { |
| 142 | + if (!active_ || !safe_) return; |
| 143 | + if (end < observed_end_ || observed_end_ < base) { |
| 144 | + invalidate(); // rewind, missed publication, or uncertain reset |
| 145 | + return; |
| 146 | + } |
| 147 | + while (observed_end_ < end) { |
| 148 | + const Trade& row = read(observed_end_++); |
| 149 | + ++work.update_rows; |
| 150 | + if (row.exit_time < barrier_ || (have_time_ && row.exit_time < last_time_)) { |
| 151 | + invalidate(); // a row-order barrier cannot be treated as a time filter |
| 152 | + return; |
| 153 | + } |
| 154 | + last_time_ = row.exit_time; |
| 155 | + have_time_ = true; |
| 156 | + accumulate(row); |
| 157 | + } |
| 158 | + } |
| 159 | + |
| 160 | + void open_origin(std::uint64_t incarnation, bool is_long) { |
| 161 | + if (!safe_ || !active_) return; |
| 162 | + auto found = origins_.find(incarnation); |
| 163 | + if (found == origins_.end()) { |
| 164 | + Origin row; |
| 165 | + row.is_long = is_long; |
| 166 | + // A genuinely new incarnation cannot have earlier completed |
| 167 | + // rows. An old/reactivated incarnation whose subtotal was |
| 168 | + // retired is uncertain and must replay instead of starting at 0. |
| 169 | + row.completed.valid = incarnation > high_closed_origin_; |
| 170 | + origins_.emplace(incarnation, row); |
| 171 | + } else if (found->second.is_long != is_long) { |
| 172 | + found->second.completed.valid = false; |
| 173 | + } |
| 174 | + } |
| 175 | + |
| 176 | + void retire_origin(std::uint64_t incarnation) { |
| 177 | + origins_.erase(incarnation); |
| 178 | + if (origins_.empty()) { |
| 179 | + ::pineforge::detail::clear_exposure_storage(origins_); |
| 180 | + } else if (origins_.bucket_count() / 4 > origins_.size()) { |
| 181 | + origins_.rehash(origins_.size()); |
| 182 | + } |
| 183 | + } |
| 184 | + |
| 185 | + void forget_origins() { |
| 186 | + ::pineforge::detail::clear_exposure_storage(origins_); |
| 187 | + ::pineforge::detail::clear_exposure_storage(live_cohorts_); |
| 188 | + } |
| 189 | + std::size_t resident_origins() const noexcept { return origins_.size(); } |
| 190 | + |
| 191 | + // Nonempty live maps pin their table rows. The adapter updates this |
| 192 | + // derived index at the same insert/erase/clear seams as those maps. |
| 193 | + // Querying only these rows avoids scanning resident historical cohorts |
| 194 | + // whose rosters are pinned for unrelated source bookkeeping. |
| 195 | + void live_cohort(const void* row, bool nonempty) { |
| 196 | + if (nonempty) { |
| 197 | + if (active_ && safe_) live_cohorts_.insert(row); |
| 198 | + } else { |
| 199 | + live_cohorts_.erase(row); |
| 200 | + if (live_cohorts_.empty()) |
| 201 | + ::pineforge::detail::clear_exposure_storage(live_cohorts_); |
| 202 | + else if (live_cohorts_.bucket_count() / 4 > live_cohorts_.size()) |
| 203 | + live_cohorts_.rehash(live_cohorts_.size()); |
| 204 | + } |
| 205 | + } |
| 206 | + template<class Visit> |
| 207 | + void for_each_live_cohort(Visit&& visit) const { |
| 208 | + for (const auto* row : live_cohorts_) visit(row); |
| 209 | + } |
| 210 | + |
| 211 | + void invalidate() { |
| 212 | + safe_ = false; |
| 213 | + forget_origins(); |
| 214 | + margins_.clear(); |
| 215 | + all_margins_ = {}; |
| 216 | + } |
| 217 | + |
| 218 | + template<class Select> |
| 219 | + bool ancestry(bool is_long, Select&& select, double& result) { |
| 220 | + ++work.queries; |
| 221 | + if (!safe_ || !active_) return finish(false); |
| 222 | + if (++query_generation_ == 0) { |
| 223 | + for (auto& row : origins_) row.second.visited = 0; |
| 224 | + ++query_generation_; |
| 225 | + } |
| 226 | + PositionTailExact total; |
| 227 | + select([&](std::uint64_t incarnation) { |
| 228 | + ++work.origin_visits; |
| 229 | + const auto found = origins_.find(incarnation); |
| 230 | + if (found == origins_.end()) { |
| 231 | + // No row for a fresh incarnation means an exact zero. |
| 232 | + // Missing older origins are deliberately uncertified. |
| 233 | + if (incarnation <= high_closed_origin_) total.valid = false; |
| 234 | + return; |
| 235 | + } |
| 236 | + auto& row = found->second; |
| 237 | + if (row.visited == query_generation_) return; // duplicate cohort membership |
| 238 | + row.visited = query_generation_; |
| 239 | + if (!row.completed.valid) total.valid = false; |
| 240 | + else if (row.is_long == is_long) total.merge(row.completed); |
| 241 | + }); |
| 242 | + return finish(total.read(result)); |
| 243 | + } |
| 244 | + |
| 245 | + bool margin(bool is_long, const std::string* named_entry, double& result) { |
| 246 | + ++work.queries; |
| 247 | + if (!safe_ || !active_) return finish(false); |
| 248 | + const auto side = is_long ? 1 : 0; |
| 249 | + if (!named_entry) return finish(all_margins_.side[side].read(result)); |
| 250 | + const auto offset = margins_.find(*named_entry); |
| 251 | + if (!offset) { |
| 252 | + result = 0.0; |
| 253 | + return finish(true); |
| 254 | + } |
| 255 | + return finish(margins_.get(*offset).side[side].read(result)); |
| 256 | + } |
| 257 | + |
| 258 | +private: |
| 259 | + bool finish(bool certified) noexcept { |
| 260 | + if (certified) ++work.certified_queries; |
| 261 | + else ++work.fallback_queries; |
| 262 | + return certified; |
| 263 | + } |
| 264 | + |
| 265 | + void accumulate(const Trade& trade) { |
| 266 | + high_closed_origin_ = std::max(high_closed_origin_, trade.entry_incarnation); |
| 267 | + const auto origin = origins_.find(trade.entry_incarnation); |
| 268 | + if (origin != origins_.end()) { |
| 269 | + if (origin->second.is_long != trade.is_long) |
| 270 | + origin->second.completed.valid = false; |
| 271 | + origin->second.completed.add(trade.qty); |
| 272 | + } |
| 273 | + if (trade.exit_id != kMarginCallLabel) return; |
| 274 | + const auto side = trade.is_long ? 1 : 0; |
| 275 | + all_margins_.side[side].add(trade.qty); |
| 276 | + const auto inserted = margins_.insert(trade.entry_id, Margins{}); |
| 277 | + auto value = margins_.get(inserted.first); |
| 278 | + value.side[side].add(trade.qty); |
| 279 | + margins_.set(inserted.first, value); |
| 280 | + } |
| 281 | + |
| 282 | + Origins origins_; |
| 283 | + std::unordered_set<const void*> live_cohorts_; |
| 284 | + // Historical margin ids must stay queryable even after their last live |
| 285 | + // origin drains. Only fixed tree metadata lives in RAM; values/ids are |
| 286 | + // in the existing external table, without a resident directory/cache. |
| 287 | + ::pineforge::detail::DiskIdTable<Margins> margins_; |
| 288 | + Margins all_margins_{}; |
| 289 | + std::uint64_t high_closed_origin_ = 0; |
| 290 | + std::size_t observed_end_ = 0; |
| 291 | + std::int64_t barrier_ = 0, last_time_ = 0; |
| 292 | + std::uint32_t query_generation_ = 0; |
| 293 | + bool active_ = false, safe_ = false, have_time_ = false; |
| 294 | +}; |
| 295 | + |
| 296 | +// The fallback is deliberately the old ordered fold over the row source. |
| 297 | +// The predicate and the additions have the same evaluation/order as L18: |
| 298 | +// resident newest-to-oldest, then external newest-to-oldest, first strict |
| 299 | +// timestamp barrier terminates the entire walk. It remains O(history). |
| 300 | +template<class Read, class Include> |
| 301 | +double replay_position_tail(std::size_t end, std::size_t base, Read&& read, |
| 302 | + const PositionTailSpill* spill, std::int64_t barrier, |
| 303 | + Include&& include, const std::string& from_entry, |
| 304 | + PineExecutionAdapter::PositionTailWork& work) { |
| 305 | + double result = 0.0; |
| 306 | + for (std::size_t index = end; index > base;) { |
| 307 | + const Trade& trade = read(--index); |
| 308 | + ++work.replay_rows; |
| 309 | + if (trade.exit_time < barrier) return result; |
| 310 | + if (include(trade.is_long, trade.entry_incarnation, trade.exit_id == kMarginCallLabel, |
| 311 | + [&] { return trade.entry_id == from_entry; })) result += trade.qty; |
| 312 | + } |
| 313 | + if (spill) { |
| 314 | + auto cursor = spill->size_bytes(); |
| 315 | + PositionTailSpill::Row trade; |
| 316 | + while (spill->previous(cursor, trade)) { |
| 317 | + ++work.replay_rows; |
| 318 | + if (trade.exit_time < barrier) break; |
| 319 | + if (include(trade.is_long, trade.entry_incarnation, trade.margin_call, |
| 320 | + [&] { return spill->entry_id_equals(trade, from_entry); })) result += trade.qty; |
| 321 | + } |
| 322 | + } |
| 323 | + return result; |
| 324 | +} |
| 325 | + |
| 326 | +} // namespace pineforge::source::detail |
0 commit comments