diff --git a/crates/moon-core/src/db/analytics/groups.rs b/crates/moon-core/src/db/analytics/groups.rs index 9915e6181..31c3b9019 100644 --- a/crates/moon-core/src/db/analytics/groups.rs +++ b/crates/moon-core/src/db/analytics/groups.rs @@ -361,6 +361,25 @@ pub(super) fn groups( by_strategy: bool, ) -> ReadResult> { const CTX: &str = "analytics: groups"; + // Keep the projection narrow: materializing the wide report source copies columns this + // aggregate never reads. The fence prevents SQLite from expanding pnl and identity CASEs + // again for every aggregate and GROUP BY expression. + let narrow = "o.strategyid, o.core_uid, o.core_name, o.coin, o.pnl, + o.profitbtc, o.spentbtc, o.basecurrency"; + let (normalized, src, raw_src) = if by_strategy { + let raw = raw_src + .map(|source| { + format!("raw_normalized AS MATERIALIZED (SELECT {narrow} FROM {source}), ") + }) + .unwrap_or_default(); + ( + format!("normalized AS MATERIALIZED (SELECT {narrow} FROM {src}), {raw}"), + "normalized o", + raw_src.map(|_| "raw_normalized o"), + ) + } else { + (String::new(), src, raw_src) + }; // Strategy key is `id@core_uid`, splitting PER CORE so each core's activity is visible. // Renames do not create groups, distinct strategies sharing a name do not merge, and the // name is only a label. @@ -458,7 +477,7 @@ pub(super) fn groups( ) }; let sql = format!( - "WITH a AS ( + "WITH {normalized}a AS ( SELECT {key} AS k, {pair} MAX(o.core_name) AS core_name, diff --git a/crates/moon-core/src/db/analytics/groups/tests.rs b/crates/moon-core/src/db/analytics/groups/tests.rs index 67a85df8c..a6efccf2a 100644 --- a/crates/moon-core/src/db/analytics/groups/tests.rs +++ b/crates/moon-core/src/db/analytics/groups/tests.rs @@ -12,6 +12,125 @@ use super::*; use crate::db::{ProfitMetric, SideFilter}; +/// The normalization fence must preserve liquidation attribution, deleted metadata and quote splits. +/// Flattening the source used to expand its identity and pnl expressions for each aggregate. +#[test] +fn normalized_strategy_groups_match_the_unfenced_reference() { + use crate::db::test_support::{build_replica, remove_db, temp_db}; + let path = temp_db("normalized-strategy-parity"); + let conn = build_replica(&path, &[(10_000, 3.0, "XALPHA"), (11_000, -2.0, "XALPHA")]); + conn.execute_batch( + "ALTER TABLE orders_rep ADD COLUMN channelname TEXT; + ALTER TABLE orders_rep ADD COLUMN signaltype TEXT; + UPDATE orders_rep SET core_name='CORE_A'; + UPDATE orders_rep SET channelname='LIQUIDATION', signaltype='UNIT_01', strategyid=0 WHERE newrecid=2; + ATTACH ':memory:' AS strat; + CREATE TABLE strat.strategies(core_uid INTEGER, strategy_id INTEGER, name TEXT, + deleted INTEGER, checked INTEGER); + CREATE TABLE strat.strategy_versions(core_uid INTEGER, strategy_id INTEGER, + valid_from INTEGER, valid_to INTEGER, raw_json TEXT); + INSERT INTO strat.strategies VALUES(1,7,'UNIT_01',1,0);", + ) + .expect("synthetic strategy fixture"); + conn.execute_batch( + "ALTER TABLE orders_rep RENAME COLUMN strategyid TO original_strategyid; + ALTER TABLE orders_rep ADD COLUMN strategyid; + UPDATE orders_rep SET strategyid=original_strategyid; + INSERT INTO orders_rep(core_uid,core_name,newrecid,closedate,buydate,profitbtc, + spentbtc,coin,strategyid,isshort,emulator,deleted,basecurrency) + VALUES(1,'CORE_A',3,12000,11700,4,200,'XALPHA','7',0,0,0,1);", + ) + .unwrap(); + assert_eq!( + conn.query_row( + "SELECT typeof(strategyid) FROM orders_rep WHERE newrecid=3", + [], + |r| r.get::<_, String>(0) + ) + .unwrap(), + "text" + ); + for split in [false, true] { + conn.execute( + "UPDATE orders_rep SET basecurrency=?1 WHERE newrecid=2", + [if split { 8 } else { 1 }], + ) + .expect("quote scope"); + { + for metric in [ProfitMetric::Quote, ProfitMetric::Percent] { + let query = Query { + axis: crate::db::ReportAxis::from_measured( + std::collections::HashMap::from([( + 1, + vec![crate::db::OffsetSegment { + from_utc: 0, + offset_secs: 3600, + }], + )]), + chrono_tz::UTC, + ), + from: -1, + to: 20_000, + metric, + prefer_usdt: true, + ..Default::default() + }; + let mut query = query; + query.floor_all_history(); + // Native is intentional for split money: scalar callers refuse that scope, + // while grouping must retain its exact per-group unknown/raw-money semantics. + let projection = if metric == ProfitMetric::Percent { + super::super::ProjectionMode::Percent + } else { + super::super::ProjectionMode::Native + }; + let src = unified_from_mode(&conn, &query, projection) + .unwrap() + .unwrap(); + let raw_src = raw_source(&conn, &query).unwrap(); + let reference = reference_with_raw(&src, raw_src.as_deref()); + let old = run(&conn, &reference, &query, true, true); + let new = groups(&conn, &src, raw_src.as_deref(), &query, true, true).unwrap(); + assert!(!old.is_empty()); + assert_same("normalization parity", &old, &new); + if !split { + let integer = new.iter().find(|g| g.key == "7@1").unwrap(); + assert_eq!(integer.n, 2); + assert_eq!(integer.raw_profit, 7.0); + assert_eq!(integer.avg_order, 150.0); + assert_eq!(new.iter().map(|g| g.raw_profit).sum::(), 5.0); + } + } + } + } + drop(conn); + remove_db(&path); +} + +/// Keep the old active aggregation while independently reading native money for Percent groups. +fn reference_with_raw(src: &str, raw_src: Option<&str>) -> String { + let sql = reference_sql(src, true, true); + let Some(raw_src) = raw_src else { + return sql; + }; + let correlated = format!( + "FROM {raw_src} WHERE CAST(o.strategyid AS TEXT) || '@' || CAST(o.core_uid AS TEXT) = active.k" + ); + format!( + "SELECT active.k, active.name, active.kind, active.core_name, active.cores_n, + active.alive, active.n, active.profit, active.wins, active.wsum, active.lsum, + active.best, active.worst, active.lastedit, active.blacklist, active.whitelist, + (SELECT COALESCE(SUM(o.profitbtc),0) {correlated}), + (SELECT COALESCE(AVG(CASE WHEN o.spentbtc>0 THEN o.spentbtc END),0) {correlated}), + (SELECT MIN(o.basecurrency) {correlated}), (SELECT MAX(o.basecurrency) {correlated}), + (SELECT COUNT(CASE WHEN typeof(o.basecurrency)='integer' THEN 1 END) {correlated}), + (SELECT COUNT(*) {correlated}) + FROM (WITH ref(k,name,kind,core_name,cores_n,alive,n,profit,wins,wsum,lsum,best,worst, + lastedit,blacklist,whitelist,raw_profit,avg_order,qmin,qmax,qint,qrows) AS ({sql}) + SELECT * FROM ref) active ORDER BY active.profit DESC, active.k" + ) +} + /// The report scope every case here uses: all cores, all history, no filters. fn q() -> Query { Query { diff --git a/crates/moon-core/src/db/analytics/mod.rs b/crates/moon-core/src/db/analytics/mod.rs index d75c0b0ec..cff9d2960 100644 --- a/crates/moon-core/src/db/analytics/mod.rs +++ b/crates/moon-core/src/db/analytics/mod.rs @@ -423,13 +423,22 @@ enum ScopeDecision { /// Errors: /// Returns a classified report read failure when quote or valuation preflight cannot complete. fn scope_decision_on(conn: &Connection, q: &Query) -> ReadResult { + scope_decision_with_quote_reader_on(conn, q, quote_breakdown_on) +} + +/// Resolve the same scope contract with a caller-specific quote scan and the shared Percent exit. +fn scope_decision_with_quote_reader_on( + conn: &Connection, + q: &Query, + read_quotes: fn(&Connection, &Query) -> ReadResult, +) -> ReadResult { if q.metric == ProfitMetric::Percent { return Ok(ScopeDecision::Comparable { unit: ProfitUnit::Percent, projection: ProjectionMode::Percent, }); } - let totals = quote_breakdown_on(conn, q)?; + let totals = read_quotes(conn, q)?; Ok(match totals.scope() { QuoteScope::Empty => ScopeDecision::Empty { projection: ProjectionMode::Native, @@ -482,7 +491,12 @@ pub(in crate::db) fn usdt_projection_on( let mut money = q.clone(); money.metric = ProfitMetric::Quote; money.prefer_usdt = true; - Ok(match scope_decision_on(conn, &money)? { + Ok(usdt_projection(scope_decision_on(conn, &money)?)) +} + +/// Extract the USDT projection without repeating a resolved money-scope preflight. +fn usdt_projection(decision: ScopeDecision) -> Option { + match decision { ScopeDecision::Comparable { unit, projection } if unit == ProfitUnit::Quote(crate::db::QuoteCurrency::usdt()) => { @@ -490,7 +504,19 @@ pub(in crate::db) fn usdt_projection_on( } ScopeDecision::Empty { projection } => Some(projection), ScopeDecision::Comparable { .. } | ScopeDecision::Split(_) => None, - }) + } +} + +/// Resolve active and overlay projections for a Quote+USDT request with identical scope inputs. +/// +/// Decisions remain local to the caller's pinned snapshot, including the native retry. +pub(in crate::db) fn projection_with_usdt_on( + conn: &Connection, + q: &Query, +) -> ReadResult<(ProjectionMode, Option)> { + let decision = scope_decision_on(conn, q)?; + let active = decision.projection().ok_or(ReadFail::IncomparableQuote)?; + Ok((active, usdt_projection(decision))) } /// Resolve one query to a safe row-level projection for non-`ProfitScope` consumers. diff --git a/crates/moon-core/src/db/analytics/profit_monitor.rs b/crates/moon-core/src/db/analytics/profit_monitor.rs index d0949c9b8..d807cb774 100644 --- a/crates/moon-core/src/db/analytics/profit_monitor.rs +++ b/crates/moon-core/src/db/analytics/profit_monitor.rs @@ -5,7 +5,10 @@ mod tests; use rusqlite::Connection; -use super::{Query, ScopeDecision, min_closedate, scope_decision_on, scoped, unified_from_mode}; +use super::{ + Query, ScopeDecision, min_closedate, scope_decision_with_quote_reader_on, scoped, + unified_from_mode, +}; use crate::db::read_fail::read_fail_on; use crate::db::{ProfitScope, QuoteCurrency, ReadFail, ReadResult}; @@ -165,7 +168,11 @@ pub(super) fn profit_monitor_on( if q.from < 0 { q.from = min_closedate(conn, &q.resolved_axis(conn)?)?; } - let decision = match scope_decision_on(conn, &q)? { + let decision = match scope_decision_with_quote_reader_on( + conn, + &q, + super::query::quote_breakdown_unfenced_on, + )? { ScopeDecision::Split(totals) => return Ok(ProfitScope::Split(totals)), decision => decision, }; diff --git a/crates/moon-core/src/db/analytics/query/mod.rs b/crates/moon-core/src/db/analytics/query/mod.rs index 9d18e9910..d8d0e7eeb 100644 --- a/crates/moon-core/src/db/analytics/query/mod.rs +++ b/crates/moon-core/src/db/analytics/query/mod.rs @@ -604,10 +604,27 @@ impl ProjectionMode { pub(in crate::db) fn quote_breakdown_on( conn: &Connection, q: &Query, +) -> ReadResult { + quote_breakdown_with_stage_on(conn, q, true) +} + +/// Let the monitor avoid materialization overhead while retaining identical coverage and retry. +pub(super) fn quote_breakdown_unfenced_on( + conn: &Connection, + q: &Query, +) -> ReadResult { + quote_breakdown_with_stage_on(conn, q, false) +} + +/// Apply one caller's normalization stage choice to both the primary read and native retry. +fn quote_breakdown_with_stage_on( + conn: &Connection, + q: &Query, + materialize: bool, ) -> ReadResult { let sources = super::super::read_sources_res(conn)?; let valuation_attached = super::super::valuation::is_attached(conn); - match quote_breakdown_attempt(conn, q, &sources, valuation_attached) { + match quote_breakdown_attempt(conn, q, &sources, valuation_attached, materialize) { Ok(totals) => Ok(totals), Err(error) if valuation_attached @@ -617,7 +634,7 @@ pub(in crate::db) fn quote_breakdown_on( &format!("DETACH DATABASE {}", super::super::valuation::SCHEMA), [], ); - quote_breakdown_attempt(conn, q, &sources, false).map_err(|retry_error| { + quote_breakdown_attempt(conn, q, &sources, false, materialize).map_err(|retry_error| { super::super::read_fail::read_fail( "analytics: quote breakdown native retry", retry_error, @@ -640,6 +657,7 @@ pub(in crate::db) fn quote_breakdown_on( /// sources: Physical report sources discovered from `main`. /// include_valuation: Whether the historical mode may join the attached derived cache; the /// current-rate mode does not depend on it. +/// materialize: Whether to fence the normalization stage for this caller. /// /// Returns: /// Exact quote totals, optionally carrying complete USDT coverage. @@ -651,6 +669,7 @@ fn quote_breakdown_attempt( q: &Query, sources: &[super::super::ReadSource], include_valuation: bool, + materialize: bool, ) -> rusqlite::Result { let has_names = strategies_attached(conn); // Resolved ONCE for this whole pass, above the source loop: the rows pass and this totals pass @@ -676,14 +695,16 @@ fn quote_breakdown_attempt( let attribution = liquidation_attribution_available(&src.cols, has_names); let where_branches = q.where_branches(&period, &src.cols, &sid, Some("r"), attribution, &mask); - let (quote, group_by) = super::super::quote::trusted_quote_group("r", &src.cols); let source = super::super::report_read::source_partition(src); - let valuation = super::super::valuation::projection( + let inputs = super::super::valuation::CoverageInputs::normalized("r"); + let names = super::super::valuation::CoverageInputs::normalized(""); + let valuation = super::super::valuation::projection_with_inputs( q.valuation, include_valuation, "r", &src.cols, source, + &inputs, ); let joins = valuation .as_ref() @@ -693,16 +714,46 @@ fn quote_breakdown_attempt( .as_ref() .map(|parts| format!(", {}", parts.aggregate_columns())) .unwrap_or_default(); + // Keep raw cells beside corrected amounts for storage and cache-identity guards. The + // monitor leaves the stage flattenable; other callers reuse the fenced normalization. + let raw_columns = super::super::valuation::RAW_COVERAGE_COLUMNS + .iter() + .copied() + .filter(|column| src.cols.contains(*column)) + .map(|column| format!("r.\"{column}\"")) + .collect::>() + .join(", "); + let normalized = super::super::valuation::CoverageInputs::report("r", &src.cols); + let group_by = if src.cols.contains("basecurrency") { + format!(" GROUP BY {}", inputs.quote) + } else { + String::new() + }; let sql = where_branches .iter() .map(|where_sql| { format!( - "SELECT {quote}, COALESCE(SUM({settled_profit}), 0.0), COUNT(*)\ - {coverage_columns} \ - FROM {} r{joins} WHERE {where_sql}{group_by}", + "SELECT * FROM (WITH normalized AS {stage} ( + SELECT {raw_columns}, {} AS {quote_name}, + {} AS {profit_name}, {} AS {spent_name} + FROM {} r WHERE {where_sql}) + SELECT {quote}, COALESCE(SUM({profit}), 0.0), COUNT(*) + {coverage_columns} + FROM normalized r{joins}{group_by})", + normalized.quote, + normalized.profit, + normalized.spent, src.table, - settled_profit = - super::super::quote::settled_amount_expr("r", &src.cols, "profitbtc"), + quote_name = names.quote, + profit_name = names.profit, + spent_name = names.spent, + quote = inputs.quote, + profit = inputs.profit, + stage = if materialize { + "MATERIALIZED" + } else { + "NOT MATERIALIZED" + }, ) }) .collect::>() diff --git a/crates/moon-core/src/db/analytics/tests.rs b/crates/moon-core/src/db/analytics/tests.rs index 0ac4f3c42..f83a4ffd3 100644 --- a/crates/moon-core/src/db/analytics/tests.rs +++ b/crates/moon-core/src/db/analytics/tests.rs @@ -1,11 +1,226 @@ //! Summary, Strategies, and quote-scope analytics regression tests. +mod real_data; + use super::super::test_support::{ build_replica, build_replica_multi_core, corrupt_leaf_page, remove_db, spread_rows, temp_db, }; use super::super::{QuoteCurrency, SideFilter}; use super::*; +/// Frozen pre-normalization quote aggregate, independently retaining the original SQL shape. +/// +/// The oracle exercises the public coverage builder; the new path consumes explicit normalized +/// expressions instead. Comparing complete results catches stale-cache and unknown-quote drift. +fn unfenced_quote_breakdown(conn: &Connection, q: &Query) -> QuoteBreakdown { + use rusqlite::types::Value; + let sources = super::super::read_sources_res(conn).unwrap(); + let mask = query::StrategyMask::resolve(conn, q).unwrap(); + let axis = q.resolved_axis_sql(conn).unwrap(); + let period = q.period_predicate_on(&axis, "r"); + let attached = super::super::valuation::is_attached(conn); + let mut groups = Vec::new(); + let mut coverage = super::super::valuation::CoverageAggregate::default(); + for src in sources { + if !src.cols.contains("closedate") || !src.cols.contains("profitbtc") { + continue; + } + let sid = effective_sid_expr("r", &src.cols, strategies_attached(conn)); + let attribution = + query::liquidation_attribution_available(&src.cols, strategies_attached(conn)); + let (quote, group_by) = super::super::quote::trusted_quote_group("r", &src.cols); + let valuation = super::super::valuation::projection( + q.valuation, + attached, + "r", + &src.cols, + super::super::report_read::source_partition(&src), + ); + let joins = valuation.as_ref().map(|v| v.joins.as_str()).unwrap_or(""); + let columns = valuation + .as_ref() + .map(|v| format!(", {}", v.aggregate_columns())) + .unwrap_or_default(); + let sql = q + .where_branches(&period, &src.cols, &sid, Some("r"), attribution, &mask) + .iter() + .map(|predicate| { + format!( + "SELECT {quote}, COALESCE(SUM({}),0.0), COUNT(*) {columns} + FROM {} r{joins} WHERE {predicate}{group_by}", + super::super::quote::settled_amount_expr("r", &src.cols, "profitbtc"), + src.table + ) + }) + .collect::>() + .join(" UNION ALL "); + let mut statement = conn.prepare(&sql).unwrap(); + let mut rows = statement.query(rusqlite::params![q.from, q.to]).unwrap(); + while let Some(row) = rows.next().unwrap() { + groups.push(( + super::super::quote::report_ordinal_from_value(&row.get::<_, Value>(0).unwrap()), + row.get::<_, f64>(1).unwrap(), + row.get::<_, i64>(2).unwrap(), + )); + if valuation.is_some() { + coverage.add_row(row, 3).unwrap(); + } + } + } + let totals = QuoteBreakdown::from_groups(groups); + if attached || q.valuation == super::super::valuation::ValuationMode::Current { + totals.with_valuation(coverage.finish()) + } else { + totals + } +} + +/// Computing coverage against normalized columns must preserve raw storage and cache freshness. +/// A normalized quote substituted into a raw storage check would incorrectly accept text money. +/// Both fenced Analytics reads and the flattenable monitor path must match the frozen SQL oracle. +#[test] +fn normalized_quote_breakdown_matches_unfenced_sql() { + use super::super::valuation::{ALGORITHM_VERSION, ValuationMode}; + let path = temp_db("normalized-coverage-parity"); + let conn = build_replica_multi_core( + &path, + &[ + (1, 10_000, 3.0, "XALPHA"), + (2, 11_000, -2.0, "XALPHA"), + (1, 8_000, 5.0, "XALPHA"), + (2, 12_000, 3.0, "XALPHA"), + (2, 13_000, 4.0, "XALPHA"), + (3, 1_800_000_000, 0.5, "UNI_RP"), + (3, 1_800_000_060, -0.01, "UNI_RP"), + ], + ); + conn.execute_batch( + "ALTER TABLE orders_rep ADD COLUMN channelname TEXT; + ALTER TABLE orders_rep ADD COLUMN signaltype TEXT; + ALTER TABLE orders_rep ADD COLUMN fname TEXT; + ALTER TABLE orders_rep ADD COLUMN sellreason TEXT; + ALTER TABLE orders_rep ADD COLUMN buyprice REAL; + UPDATE orders_rep SET core_name='CORE_A'; + UPDATE orders_rep SET profitbtc='3x' WHERE newrecid=4; + UPDATE orders_rep SET spentbtc='100x' WHERE newrecid=5; + UPDATE orders_rep SET fname='Pump_USD-UNI_RP_UNIT_01', buyprice=10 WHERE newrecid=6; + UPDATE orders_rep SET fname='', sellreason='LIQUIDATION', buyprice=10 WHERE newrecid=7; + UPDATE orders_rep SET channelname='LIQUIDATION', signaltype='UNIT_01', strategyid=0 WHERE newrecid=2; + ATTACH ':memory:' AS strat; + CREATE TABLE strat.strategies(core_uid INTEGER, strategy_id INTEGER, name TEXT, + deleted INTEGER, checked INTEGER); + CREATE TABLE strat.strategy_versions(core_uid INTEGER, strategy_id INTEGER, + valid_from INTEGER, valid_to INTEGER, raw_json TEXT); + INSERT INTO strat.strategies VALUES(1,7,'UNIT_01',1,0);" + ).unwrap(); + super::super::rep::core_offset::ensure_table(&conn).unwrap(); + super::super::rep::core_offset::store_segment(&conn, 1, 0, 3600, 0, "UNIT_01").unwrap(); + assert_eq!( + conn.query_row( + "SELECT typeof(profitbtc) FROM orders_rep WHERE newrecid=4", + [], + |r| r.get::<_, String>(0) + ) + .unwrap(), + "text" + ); + assert_eq!( + conn.query_row( + "SELECT typeof(spentbtc) FROM orders_rep WHERE newrecid=5", + [], + |r| r.get::<_, String>(0) + ) + .unwrap(), + "text" + ); + let sources = super::super::read_sources_res(&conn).unwrap(); + super::super::quote::learn_coin_m_cores(&conn, &sources); + let source = sources.iter().find(|s| s.table == "orders_rep").unwrap(); + let settled = super::super::quote::settled_amount_expr("r", &source.cols, "profitbtc"); + let corrected: f64 = conn + .query_row( + &format!("SELECT {settled} FROM orders_rep r WHERE newrecid=7"), + [], + |r| r.get(0), + ) + .unwrap(); + assert!((corrected + 0.1).abs() < 1e-12); + let cache_path = path.with_file_name("normalized-valuation.sqlite"); + remove_db(&cache_path); + drop(super::super::valuation::open_store(&cache_path).unwrap()); + conn.execute( + "ATTACH DATABASE ?1 AS valuation", + [cache_path.to_str().unwrap()], + ) + .unwrap(); + for quote in ["1", "8", "8.5", "'unknown'", "NULL"] { + conn.execute_batch(&format!( + "UPDATE orders_rep SET basecurrency={quote} WHERE newrecid=2" + )) + .unwrap(); + conn.execute("DELETE FROM valuation.trade_values", []) + .unwrap(); + conn.execute( + "INSERT INTO valuation.trade_values(source_kind,core_uid,row_id,algorithm_version, + closedate,quote_ordinal,profit_quote,spent_quote,rate_minute_utc,rate_usdt, + profit_usdt,spent_usdt,valued_at_ms) + SELECT 0,core_uid,newrecid,?1,closedate,basecurrency,profitbtc,spentbtc, + (closedate/60)*60,0.999,profitbtc*0.999,spentbtc*0.999,1 FROM orders_rep + WHERE typeof(basecurrency)='integer' AND basecurrency BETWEEN 0 AND 20", + [ALGORITHM_VERSION], + ) + .unwrap(); + for stale in [false, true] { + if stale { + conn.execute("UPDATE valuation.trade_values SET profit_quote=999", []) + .unwrap(); + } + for valuation in [ValuationMode::Historical, ValuationMode::Current] { + for metric in [ProfitMetric::Quote, ProfitMetric::Percent] { + for all in [false, true] { + let mut query = Query { + from: if all { -1 } else { 6000 }, + to: 2_000_000_000, + prefer_usdt: true, + metric, + valuation, + axis: super::super::ReportAxis::from_measured( + std::collections::HashMap::from([( + 1, + vec![super::super::OffsetSegment { + from_utc: 0, + offset_secs: 3600, + }], + )]), + chrono_tz::UTC, + ), + ..Default::default() + }; + query.floor_all_history(); + let old = unfenced_quote_breakdown(&conn, &query); + let new = quote_breakdown_on(&conn, &query).unwrap(); + assert_eq!(format!("{old:?}"), format!("{new:?}")); + let monitor = query::quote_breakdown_unfenced_on(&conn, &query).unwrap(); + assert_eq!(format!("{old:?}"), format!("{monitor:?}")); + assert_eq!(new.orders, if all { 7 } else { 6 }); + if !all { + conn.execute("UPDATE core_time_offset SET offset_secs=0", []) + .unwrap(); + let unshifted = quote_breakdown_on(&conn, &query).unwrap(); + assert_eq!(unshifted.orders, 7); + conn.execute("UPDATE core_time_offset SET offset_secs=3600", []) + .unwrap(); + } + } + } + } + } + } + drop(conn); + remove_db(&path); + remove_db(&cache_path); +} + /// Dividing by a drawdown displayed as zero produces a misleading recovery KPI; /// rounding the denominator before division also changes otherwise valid ratios. #[test] diff --git a/crates/moon-core/src/db/analytics/tests/real_data.rs b/crates/moon-core/src/db/analytics/tests/real_data.rs new file mode 100644 index 000000000..d9d8b9bc7 --- /dev/null +++ b/crates/moon-core/src/db/analytics/tests/real_data.rs @@ -0,0 +1,994 @@ +//! Opt-in Analytics read timings on a disposable copy, with no persisted identities in output. +//! +//! Run this test alone with --ignored --exact --nocapture --test-threads=1 in release mode. +//! MOON_ANALYTICS_BENCH_DIR is the application data root (on Windows it contains data/). +//! MOON_ANALYTICS_BENCH_SQL_DUMP optionally writes the top 30 expanded statements outside the repo. +//! The profiler and path override are process-global; never combine this probe with other tests. + +use std::collections::HashMap; +use std::path::PathBuf; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Barrier, Mutex}; +use std::time::Instant; + +use chrono::Datelike; +use rusqlite::StatementStatus; + +use super::super::{self as analytics, Query}; +use crate::config::paths; +use crate::db::trace::{self, ProfiledStatement}; +use crate::db::valuation::ValuationMode; +use crate::db::{self, ProfitMetric, ReadResult, ReportAxis, SideFilter, tuner}; + +static CAPTURED: Mutex> = Mutex::new(Vec::new()); +static CAPTURING: AtomicBool = AtomicBool::new(false); +const DAY: i64 = 86_400; + +/// Capture only action traffic, excluding discovery and the plan replay itself. +fn record(statement: ProfiledStatement) { + if CAPTURING.load(Ordering::Relaxed) { + CAPTURED.lock().expect("profiler mutex").push(statement); + } +} + +/// Drain the process-wide sink after all action threads have joined. +fn drain() -> Vec { + std::mem::take(&mut *CAPTURED.lock().expect("profiler mutex")) +} + +/// Keep read failures explicit without printing errors that can carry private SQL literals. +#[track_caller] +fn checked(label: &str, read: ReadResult) -> T { + match read { + Ok(value) => value, + Err(error) => panic!( + "benchmark read failed: {label}; variant={:?}; kind={:?}; code={:?}; private details withheld", + std::mem::discriminant(&error), + error.kind(), + error.code() + ), + } +} + +/// Replay diagnostics for one distinct expanded statement, never mixed into measured SQL time. +#[derive(Default)] +struct Plan { + class: String, + fullscan: i64, + vm: i64, + rows: usize, +} + +/// Aggregate one exact statement across actions while retaining its independent replay counters. +#[derive(Default)] +struct Statement { + ms: f64, + calls: usize, + plan: Plan, +} + +/// One invocation; SQL counts and duplicates refer to this action only. +#[derive(Clone)] +struct Sample { + wall: f64, + sql: f64, + statements: usize, + duplicates: usize, + rows: usize, + texts: Vec, + degraded: bool, +} + +/// First invocation and warm samples retained until every statement has been replayed. +struct Action { + surface: String, + scope: String, + first: f64, + warm: Sample, + attached: bool, + failures: usize, +} + +/// Optional action-specific wall time, used for the maximum simultaneous reader latency. +struct Outcome { + rows: usize, + wall: Option, +} + +impl From for Outcome { + /// Ordinary actions use the enclosing timer. + fn from(rows: usize) -> Self { + Self { rows, wall: None } + } +} + +impl From<(usize, f64)> for Outcome { + /// Contention reports the slowest synchronized reader rather than thread startup time. + fn from((rows, wall): (usize, f64)) -> Self { + Self { + rows, + wall: Some(wall), + } + } +} + +/// Measure five invocations; warm time averages the middle two of the last four samples. +/// Counts and plans describe the upper middle sample; SQL time averages the same middle pair. +fn measure>( + actions: &mut Vec, + statements: &mut HashMap, + surface: &str, + scope: &str, + mut body: impl FnMut() -> T, +) { + let attached = db::open_reader().ok().is_some_and(|conn| { + conn.query_row("SELECT count(*) FROM strat.strategies", [], |row| { + row.get::<_, i64>(0) + }) + .is_ok_and(|count| count > 0) + }); + let mut samples = Vec::new(); + let mut failures = 0; + for _ in 0..5 { + drain(); + CAPTURING.store(true, Ordering::Relaxed); + let start = Instant::now(); + let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(&mut body)); + let failed = result.is_err(); + failures += usize::from(failed); + let outcome = result + .map(Into::into) + .unwrap_or_else(|_| Outcome::from(0usize)); + let wall = outcome + .wall + .unwrap_or_else(|| start.elapsed().as_secs_f64() * 1_000.0); + CAPTURING.store(false, Ordering::Relaxed); + let captured = drain(); + let mut counts = HashMap::::new(); + let mut sql = 0.0; + for item in &captured { + let ms = item.duration.as_secs_f64() * 1_000.0; + sql += ms; + let key = if item.expanded { + item.sql.clone() + } else { + // Placeholder-only captures cannot safely be replayed. + format!("NOT_EXPANDED {}", item.sql) + }; + *counts.entry(key.clone()).or_default() += 1; + let entry = statements.entry(key).or_default(); + entry.ms += ms; + entry.calls += 1; + } + samples.push(Sample { + wall, + sql, + statements: captured.len(), + duplicates: counts + .iter() + .filter(|(sql, _)| !sql.starts_with("NOT_EXPANDED ")) + .map(|(_, n)| n.saturating_sub(1)) + .sum(), + rows: outcome.rows, + texts: counts + .into_iter() + .flat_map(|(sql, count)| std::iter::repeat_n(sql, count)) + .collect(), + degraded: failed || outcome.rows == 0 || !attached, + }); + } + let degraded = samples.iter().any(|sample| sample.degraded); + let first = samples.remove(0).wall; + samples.sort_by(|a, b| a.wall.total_cmp(&b.wall)); + let middle = samples.len() / 2; + let wall = (samples[middle - 1].wall + samples[middle].wall) / 2.0; + let sql = (samples[middle - 1].sql + samples[middle].sql) / 2.0; + let mut warm = samples.remove(middle); + warm.wall = wall; + warm.sql = sql; + warm.degraded = degraded; + actions.push(Action { + surface: surface.into(), + scope: scope.replace("m1", "CURRENT(empty-rates fallback)"), + first, + warm, + attached, + failures, + }); +} + +/// Match a plan's table or alias back to the five persistent tables under investigation. +fn target_scan(detail: &str, sql: &str) -> bool { + let Some(scanned) = detail + .strip_prefix("SCAN ") + .and_then(|rest| rest.split_whitespace().next()) + else { + return false; + }; + let tokens: Vec<_> = sql + .split(|c: char| !(c.is_ascii_alphanumeric() || c == '_' || c == '.')) + .filter(|token| !token.is_empty()) + .collect(); + [ + "orders_rep", + "closed_sell_reports", + "trade_values", + "strategies", + "strategy_versions", + ] + .iter() + .any(|table| { + if scanned.rsplit('.').next() == Some(*table) { + return true; + } + tokens.iter().enumerate().any(|(index, token)| { + if token.rsplit('.').next() != Some(*table) { + return false; + } + let mut alias = tokens.get(index + 1).copied(); + if alias.is_some_and(|word| word.eq_ignore_ascii_case("AS")) { + alias = tokens.get(index + 2).copied(); + } + alias == Some(scanned) + }) + }) +} + +/// Resolve production functions on a fresh reader, then fully step only read-only SQL. +fn replay(sql: &str) -> Plan { + let normalized = sql.split_whitespace().collect::>().join(" "); + let upper = normalized.to_ascii_uppercase(); + let strategy = [ + "FROM STRATEGIES", + "FROM STRATEGY_VERSIONS", + "JOIN STRATEGIES", + "JOIN STRATEGY_VERSIONS", + ] + .iter() + .any(|table| upper.contains(table)); + let reports = [ + "ORDERS_REP", + "CLOSED_SELL_REPORTS", + "TRADE_VALUES", + "STRAT.", + "VALUATION.", + ] + .iter() + .any(|table| upper.contains(table)); + if !upper.starts_with("SELECT ") || (!strategy && !reports) || sql.starts_with("NOT_EXPANDED ") + { + return Plan { + class: "UNSUPPORTED".into(), + ..Default::default() + }; + } + let attempt = || -> Option { + let reader; + let strategy_reader; + let conn = if strategy && !reports { + strategy_reader = rusqlite::Connection::open_with_flags( + paths::strategies_db_path(), + rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY, + ) + .ok()?; + &strategy_reader + } else { + reader = db::open_reader().ok()?; + &*reader + }; + if !strategy || reports { + let axis = ReportAxis::load(conn, chrono_tz::UTC).ok()?; + analytics::time_zone::install(conn, &axis).ok()?; + } + let mut statement = conn.prepare(sql).ok()?; + if !statement.readonly() { + return Some(Plan { + class: "NOT_READONLY".into(), + ..Default::default() + }); + } + let mut explain = conn.prepare(&format!("EXPLAIN QUERY PLAN {sql}")).ok()?; + let details = explain + .query_map([], |row| row.get::<_, String>(3)) + .ok()? + .collect::, _>>() + .ok()?; + let scan = details.iter().any(|line| target_scan(line, sql)); + let search = details.iter().any(|line| line.contains("SEARCH")); + let temp = details.iter().any(|line| line.contains("TEMP B-TREE")); + let class = [ + scan.then_some("SCAN"), + search.then_some("SEARCH"), + temp.then_some("TEMP_B-TREE"), + ] + .into_iter() + .flatten() + .collect::>() + .join("+"); + let mut count = 0; + { + let mut rows = statement.query([]).ok()?; + while rows.next().ok()?.is_some() { + count += 1; + } + } + Some(Plan { + class: if class.is_empty() { + "OTHER".into() + } else { + class + }, + fullscan: i64::from(statement.get_status(StatementStatus::FullscanStep)), + vm: i64::from(statement.get_status(StatementStatus::VmStep)), + rows: count, + }) + }; + attempt().unwrap_or_else(|| Plan { + class: "UNSUPPORTED".into(), + ..Default::default() + }) +} + +/// Remove all literal values and comments before exposing a SQL prefix in public output. +fn public_sql(sql: &str) -> String { + let mut chars = sql.chars().peekable(); + let mut out = String::new(); + while let Some(ch) = chars.next() { + if ch == '\'' || ch == '"' || ch == '`' || ch == '[' { + let end = if ch == '[' { ']' } else { ch }; + out.push('?'); + while let Some(next) = chars.next() { + if next == end { + if chars.peek() == Some(&end) { + chars.next(); + } else { + break; + } + } + } + } else if ch.is_ascii_digit() { + out.push('?'); + while chars + .peek() + .is_some_and(|c| c.is_ascii_digit() || *c == '.') + { + chars.next(); + } + } else if ch == '-' && chars.peek() == Some(&'-') { + chars.next(); + for next in chars.by_ref() { + if next == '\n' { + break; + } + } + } else if ch == '/' && chars.peek() == Some(&'*') { + chars.next(); + while let Some(next) = chars.next() { + if next == '*' && chars.peek() == Some(&'/') { + chars.next(); + break; + } + } + } else { + out.push(if ch.is_ascii() && !ch.is_control() { + ch + } else { + ' ' + }); + } + } + out.split_whitespace() + .collect::>() + .join(" ") + .chars() + .take(120) + .collect() +} + +/// Discovery identities stay in memory; labels use only their rank by report row count. +struct Pool { + cores: Vec, + strategies: Vec<(i64, Option)>, + coins: Vec, +} + +/// Discover the largest cores, strategies and five coins without printing persisted names. +fn discover() -> Pool { + let conn = checked("discover.open_reader", db::open_reader()); + let cores = conn.prepare("SELECT core_uid FROM orders_rep GROUP BY core_uid ORDER BY count(*) DESC, core_uid LIMIT 5") + .expect("core discovery").query_map([], |row| Ok(row.get::<_, i64>(0)? as u64)) + .expect("core discovery").collect::, _>>().expect("core discovery"); + let strategies = conn.prepare("SELECT strategyid, core_uid FROM orders_rep WHERE strategyid != 0 GROUP BY strategyid, core_uid ORDER BY count(*) DESC, core_uid, strategyid") + .expect("strategy discovery").query_map([], |row| Ok((row.get(0)?, Some(row.get::<_, i64>(1)? as u64)))) + .expect("strategy discovery").collect::, _>>().expect("strategy discovery"); + let coins = conn.prepare("SELECT coin FROM orders_rep WHERE coin IS NOT NULL GROUP BY coin ORDER BY count(*) DESC, coin LIMIT 5") + .expect("coin discovery").query_map([], |row| row.get(0)) + .expect("coin discovery").collect::, _>>().expect("coin discovery"); + Pool { + cores, + strategies, + coins, + } +} + +/// Count rendered summary points, including honest split scopes without scalar payloads. +fn summary_rows(q: &Query, cores: bool) -> usize { + let result = analytics::summary_data(q, cores); + checked("summary_data.undated", result.undated); + if let Some(cores) = result.cores { + checked("summary_data.cores", cores); + } + checked("summary_data.data", result.data) + .data() + .map_or(0, |s| s.days.len()) +} + +/// Count strategy rows and check every independent result from the compound surface. +fn strategy_rows(q: &Query) -> usize { + let result = analytics::strategy_base_data(q, true); + checked("strategy_base_data.undated", result.undated); + if let Some(cores) = result.cores { + checked("strategy_base_data.cores", cores); + } + checked("strategy_base_data.data", result.data) + .data() + .map_or(0, |s| s.strategies.len()) +} + +/// Build one period/metric scope with the actual persisted core-clock correction. +fn query(to: i64, days: i64, metric: usize, axis: &ReportAxis) -> Query { + Query { + from: if days == 0 { -1 } else { to - days * DAY }, + to, + axis: axis.clone(), + metric: if metric == 2 { + ProfitMetric::Percent + } else { + ProfitMetric::Quote + }, + prefer_usdt: metric != 2, + valuation: if metric == 1 { + ValuationMode::Current + } else { + ValuationMode::Historical + }, + ..Default::default() + } +} + +/// Exercise every Batch 1 surface without running the app or accessing a default data root. +#[test] +#[ignore = "requires a disposable data root in MOON_ANALYTICS_BENCH_DIR; run alone"] +fn real_data_read_bench() { + // Windows canonicalization adds a verbatim prefix that SQLite file URIs cannot attach. + let root = std::path::absolute(PathBuf::from( + std::env::var_os("MOON_ANALYTICS_BENCH_DIR").expect("MOON_ANALYTICS_BENCH_DIR is required"), + )) + .expect("copy root must be absolute"); + assert!(root.is_dir(), "copy root must exist"); + assert!( + paths::set_data_dir_override(root.clone()), + "run this test alone" + ); + // Check the side-effect-free path first: reports_db_path may migrate a flat root. + assert!(paths::db_dir_path().starts_with(&root)); + assert!( + paths::db_dir_path().is_dir(), + "copy requires the platform database subdirectory; no files moved" + ); + assert!(paths::reports_db_path().starts_with(&root)); + assert!( + paths::reports_db_path().is_file(), + "copy reports database is missing" + ); + let _permit = db::report_recovery::prepare().expect("copy recovery lease unavailable"); + if std::env::var("MOON_ANALYTICS_BENCH_ANALYZE").as_deref() == Ok("1") { + for path in [ + paths::reports_db_path(), + paths::strategies_db_path(), + paths::valuation_db_path(), + ] { + assert!(path.starts_with(&root), "ANALYZE must stay on the copy"); + let conn = rusqlite::Connection::open_with_flags( + path, + rusqlite::OpenFlags::SQLITE_OPEN_READ_WRITE, + ) + .expect("copy ANALYZE connection"); + conn.execute_batch("ANALYZE").expect("ANALYZE on copy"); + } + } + let pool = discover(); + assert!( + !pool.cores.is_empty() && !pool.strategies.is_empty(), + "copy must contain cores and strategies" + ); + let conn = checked("setup.open_reader", db::open_reader()); + let axis = checked("setup.report_axis", ReportAxis::load(&conn, chrono_tz::UTC)); + drop(conn); + let now = chrono::DateTime::from_timestamp(crate::util::now_unix_ms_i64().div_euclid(1000), 0) + .expect("current timestamp"); + let to = now.timestamp().div_euclid(DAY) * DAY + DAY; + assert!( + trace::install_read_profiler(record), + "run this test in a fresh process" + ); + let mut actions = Vec::new(); + let mut statements = HashMap::new(); + macro_rules! bench { + ($name:expr, $scope:expr, $body:expr) => { + measure(&mut actions, &mut statements, $name, $scope, $body) + }; + } + println!( + "N=5; FIRST is process-order first, not OS cold; WARM is median of four; periods end at next UTC midnight." + ); + println!( + "ROWS counts returned payload rows; split scopes can return zero scalar rows. Counts/plans use upper middle sample. SQL_MS is PROFILE time (1ms resolution); contention sums overlap." + ); + bench!("open_reader", "all", || { + checked("open_reader", db::open_reader()); + 0 + }); + let basis_reader = checked("basis_probe.open_reader", db::open_reader()); + bench!("basis_probe", "test-only/uncached", || checked( + "basis_probe", + analytics::basis::margin_cores(&basis_reader) + ) + .len()); + let probe = actions.last().expect("basis measurement").warm.clone(); + drop(basis_reader); + for days in [1, 30, 365, 0] { + for metric in 0..3 { + let q = query(to, days, metric, &axis); + for read_cores in [false, true] { + bench!( + "summary_data", + &format!("D{days}/m{metric}/cores={read_cores}"), + || summary_rows(&q, read_cores) + ); + } + } + } + for metric in 0..3 { + for (label, cores) in [ + ("all", vec![]), + ("core#1", pool.cores[..1].to_vec()), + ("C5", pool.cores.clone()), + ] { + for (side_label, side) in [ + ("all", SideFilter::All), + ("long", SideFilter::Long), + ("short", SideFilter::Short), + ] { + let mut q = query(to, 365, metric, &axis); + q.cores = cores.clone(); + q.side = side; + for read_cores in [false, true] { + bench!( + "summary_data", + &format!("D365/m{metric}/{label}/{side_label}/cores={read_cores}"), + || summary_rows(&q, read_cores) + ); + } + } + } + for days in [365, 0] { + for one in [false, true] { + let mut q = query(to, days, metric, &axis); + if one { + q.cores = pool.cores[..1].to_vec(); + } + bench!( + "strategy_base_data", + &format!("D{days}/m{metric}/core#1={one}"), + || strategy_rows(&q) + ); + } + } + for days in [30, 365, 1] { + let mut q = query(to, days, metric, &axis); + if days == 30 { + q.from = now + .date_naive() + .with_day(1) + .expect("month start") + .and_hms_opt(0, 0, 0) + .expect("midnight") + .and_utc() + .timestamp(); + } + let mut previous = q.clone(); + previous.to = q.from; + previous.from -= q.to - q.from; + bench!( + "calendar_data", + &format!( + "{}/m{metric}/comparison/hourly={}", + if days == 30 { + "MONTH".to_string() + } else { + format!("D{days}") + }, + days == 1 + ), + || { + let data = analytics::calendar_data(&q, Some(&previous), days == 1, true); + checked("calendar_data.undated", data.undated); + if let Some(cores) = data.cores { + checked("calendar_data.cores", cores); + } + checked("calendar_data.period", data.period) + .data() + .map_or(0, |period| period.current.len()) + } + ); + let raw = actions.last().expect("Calendar measurement"); + let probe_queries = probe + .texts + .iter() + .filter(|sql| sql.contains("spentbtc * lev / (boughtq * buyprice)")) + .count(); + let calendar_queries = raw + .warm + .texts + .iter() + .filter(|sql| sql.contains("spentbtc * lev / (boughtq * buyprice)")) + .count(); + let calls = if probe_queries == 0 { + 0.0 + } else { + calendar_queries as f64 / probe_queries as f64 + }; + let mut warm = raw.warm.clone(); + warm.wall = (warm.wall - calls * probe.wall).max(0.0); + warm.sql = (warm.sql - calls * probe.sql).max(0.0); + actions.push(Action { + surface: "calendar_production_cache_ESTIMATE".into(), + scope: raw.scope.clone(), + first: (raw.first - calls * probe.wall).max(0.0), + warm, + attached: raw.attached, + failures: raw.failures, + }); + } + } + let base = query(to, 365, 0, &axis); + bench!("undated_closes", "D365", || { + checked("undated_closes", analytics::undated_closes(&base)); + 1 + }); + let variants = [tuner::Variant::default()]; + let field = tuner::FIELDS[0].col; + for metric in [0, 2] { + for one in [false, true] { + let mut q = query(to, 365, metric, &axis); + if one { + q.strategies = pool.strategies[..1].to_vec(); + } + let scope = format!("D365/m{metric}/strat#1={one}"); + bench!("filter_tuner_data", &scope, || { + let data = tuner::filter_tuner_data(&q, &variants, field, 40); + checked("filter_tuner_data.stats", data.stats); + checked("filter_tuner_data.histogram", data.histogram).len() + }); + bench!("variant_stats", &scope, || checked( + "variant_stats", + tuner::variant_stats(&q, &variants) + ) + .len()); + bench!("histogram", &scope, || checked( + "histogram", + tuner::histogram(&q, field, 40) + ) + .len()); + bench!("suggest_field", &scope, || usize::from( + checked( + "suggest_field", + tuner::suggest_field(&q, field, 10, 40, true) + ) + .is_some() + )); + if metric == 0 { + bench!("time_tuner_data", &scope, || { + let data = tuner::time_tuner_data(&q, &variants); + checked("time_tuner_data.stats", data.stats); + checked("time_tuner_data.slider", data.slider); + checked("time_tuner_data.profiles", data.profiles).len() + }); + bench!("suggest_time", &scope, || { + checked( + "suggest_time", + tuner::suggest_time( + &q, + 10, + 40, + true, + tuner::TimeAxes { + week: true, + day: true, + hour: true, + ..Default::default() + }, + ), + ); + 1 + }); + bench!("coin_tuner_data", &scope, || { + let data = tuner::coin_tuner_data(&q, &variants, &base, &pool.coins); + checked("coin_tuner_data.kpi", data.kpi); + checked("coin_tuner_data.picked_strategies", data.picked_strategies); + checked("coin_tuner_data.stats", data.stats).len() + }); + bench!("strategies_for_coins", &scope, || checked( + "strategies_for_coins", + analytics::strategies_for_coins(&q, &pool.coins) + ) + .len()); + } + } + } + for (label, strategies) in [ + ("strat#1", pool.strategies[..1].to_vec()), + ( + "S50", + pool.strategies[..pool.strategies.len().min(50)].to_vec(), + ), + ("all", vec![]), + ] { + let mut q = base.clone(); + q.strategies = strategies; + bench!("read_deals", label, || checked( + "read_deals", + tuner::ticks::read_deals(&q) + ) + .deals + .len()); + } + let pairs: Vec<_> = pool + .strategies + .iter() + .map(|(id, core)| (*id, core.expect("discovered core"))) + .collect(); + bench!("strategy_kinds", "all", || tuner::strategy_kinds(&pairs) + .len()); + let keys = [field.to_string()]; + bench!("live_strategies", "first-cache-miss/then-warm", || { + tuner::live_strategies(&keys).len() + }); + let param_keys = tuner::ticks::params::param_keys(); + for (label, selected) in [ + ("strat#1", pool.strategies[..1].to_vec()), + ( + "S50", + pool.strategies[..pool.strategies.len().min(50)].to_vec(), + ), + ("all", pool.strategies.clone()), + ] { + let mut q = base.clone(); + q.strategies = selected.clone(); + let deals = checked("entry_exit.read_deals", tuner::ticks::read_deals(&q)).deals; + let mut own = HashMap::new(); + bench!("entry_exit_own_values_reads", label, || { + own.clear(); + for deal in &deals { + own.entry((deal.strategy_id, Some(deal.core_uid))) + .or_insert_with(|| { + tuner::strategy_current_values( + deal.strategy_id, + Some(deal.core_uid), + ¶m_keys, + ) + }); + } + own.values().map(HashMap::len).sum::() + }); + bench!("entry_exit_now_values_missing_reads", label, || { + selected + .iter() + .filter(|key| !own.contains_key(key)) + .map(|&(id, core)| tuner::strategy_current_values(id, core, ¶m_keys).len()) + .sum::() + }); + bench!( + "strategy_values_at", + &format!("{label}/first-100-deals"), + || { + deals + .iter() + .take(100) + .filter_map(|deal| { + tuner::strategy_values_at( + deal.strategy_id, + Some(deal.core_uid), + deal.buy_ms, + ¶m_keys, + ) + }) + .map(|values| values.len()) + .sum::() + } + ); + bench!("coin_lists", label, || { + let lists = checked("coin_lists", db::coin_lists::coin_lists(&selected)); + lists.black.len() + lists.white.len() + }); + } + for days in [30, 365] { + for metric in [0, 2] { + let q = query(to, days, metric, &axis); + let scope = format!("D{days}/m{metric}"); + bench!("profit_monitor_snapshot", &scope, || { + let data = checked( + "profit_monitor_snapshot", + analytics::profit_monitor_snapshot(&q), + ); + data.scope.data().map_or(0, |s| s.cores.len()) + + data + .currencies + .iter() + .map(|c| c.data.cores.len()) + .sum::() + }); + bench!("query_totals_sliced", &scope, || { + let conn = checked("query_totals_sliced.open_reader", db::open_reader()); + let filter = db::ReportFilter { + date_from: Some(q.from), + date_to: Some(q.to - 1), + rows: db::RowScope::Closed, + axis: q.axis.clone(), + valuation: q.valuation, + ..Default::default() + }; + let slices: Vec<_> = (0..30) + .map(|i| db::TotalsSlice { + core_uids: None, + date_from: Some(to - (30 - i) * DAY), + date_to: Some(to - (29 - i) * DAY - 1), + }) + .collect(); + checked( + "query_totals_sliced", + db::query_totals_sliced(&conn, &filter, &slices), + ) + .len() + }); + } + } + let (id, core) = pool.strategies[0]; + bench!("strategy_filters", "strat#1", || { + tuner::strategy_filters(id, core, &HashMap::new()); + 1 + }); + bench!("strategy_cores", "strat#1", || tuner::strategy_cores(id) + .len()); + bench!("steady_state_repeat", "D365/summary+strategies/x10", || { + (0..10) + .map(|_| summary_rows(&base, true) + strategy_rows(&base)) + .sum::() + }); + bench!("contention", "D365/8-threads/max-wall", || { + let barrier = Barrier::new(8); + std::thread::scope(|scope| { + let handles: Vec<_> = (0..8) + .map(|_| { + scope.spawn(|| { + barrier.wait(); + let start = Instant::now(); + let rows = summary_rows(&base, true); + (start.elapsed(), rows) + }) + }) + .collect(); + let results: Vec<_> = handles + .into_iter() + .map(|h| h.join().expect("summary thread")) + .collect(); + let max = results + .iter() + .map(|(wall, _)| wall.as_secs_f64() * 1000.0) + .fold(0.0, f64::max); + (results.iter().map(|(_, rows)| rows).sum::(), max) + }) + }); + for (sql, statement) in &mut statements { + statement.plan = replay(sql); + } + actions.sort_by(|a, b| b.warm.wall.total_cmp(&a.warm.wall)); + println!( + "RANK SURFACE/ACTION SCOPE FIRST_MS WARM_MS SQL_MS NONSQL_MS STMTS DUPS WORST_PLAN FULLSCAN_STEPS ROWS STRATEGIES_ATTACHED RESULT FAILED_SAMPLES" + ); + for (rank, action) in actions.iter().enumerate() { + let plans: Vec<_> = action + .warm + .texts + .iter() + .map(|sql| &statements[sql].plan) + .collect(); + let worst = plans.iter().max_by_key(|p| { + ( + p.class.contains("SCAN"), + p.class.contains("TEMP"), + p.fullscan, + ) + }); + println!( + "{} {} {} {:.3} {:.3} {:.3} {:.3} {} {} {} {} {} {} {} {}", + rank + 1, + action.surface, + action.scope, + action.first, + action.warm.wall, + action.warm.sql, + action.warm.wall - action.warm.sql, + action.warm.statements, + action.warm.duplicates, + worst.map_or("NONE", |p| p.class.as_str()), + plans.iter().map(|p| p.fullscan).sum::(), + action.warm.rows, + action.attached, + if action.failures > 0 { + "FAILED" + } else if action.warm.degraded { + "DEGRADED" + } else { + "ROWS>0" + }, + action.failures + ); + } + let mut ranked: Vec<_> = statements.into_iter().collect(); + ranked.sort_by(|a, b| b.1.ms.total_cmp(&a.1.ms)); + if let Some(path) = std::env::var_os("MOON_ANALYTICS_BENCH_SQL_DUMP") { + let path = std::path::absolute(PathBuf::from(path)).expect("SQL dump absolute path"); + let parent = path + .parent() + .expect("SQL dump parent") + .canonicalize() + .expect("SQL dump parent exists"); + let repo = PathBuf::from(env!("CARGO_MANIFEST_DIR")) + .join("../..") + .canonicalize() + .expect("repository root"); + assert!( + !parent.starts_with(repo), + "SQL dump must be outside repository" + ); + assert!(!path.exists(), "SQL dump must be a new local file"); + let dump: Vec<_> = ranked.iter().take(30).map(|(sql, _)| sql).collect(); + let file = std::fs::OpenOptions::new() + .write(true) + .create_new(true) + .open(path) + .expect("create private SQL dump"); + serde_json::to_writer_pretty(file, &dump).expect("write private SQL dump"); + } + println!("TOP_STATEMENTS TOTAL_MS CALLS PLAN FULLSCAN_STEPS VM_STEPS REPLAY_ROWS SQL_PREFIX"); + for (sql, statement) in ranked { + println!( + "{:.3} {} {} {} {} {} {}", + statement.ms, + statement.calls, + statement.plan.class, + statement.plan.fullscan, + statement.plan.vm, + statement.plan.rows, + public_sql(&sql) + ); + } + println!( + "NOT MEASURED: run_ready_pass/count_strategies replay; resolve_addresses (UI crate); GPUI paint/layout; cx.notify rebuild" + ); + println!( + "m0=Quote+USDT/Historical CURRENT(empty-rates fallback)=offline Current m2=Percent D0=ALL; query_totals_sliced returns all native/USDT/percent totals in both metric-labelled runs." + ); + println!( + "Calendar production-cache ESTIMATE subtracts separately measured uncached basis cost per observed probe; counts/plans remain RAW. It is not a production-binary measurement." + ); + println!( + "DEGRADED means an empty payload, missing/empty strategies attachment, or a caught read failure; an empty optional result may be legitimate. Entry/Exit rows count returned parameter values; UI folding is NOT MEASURED." + ); + assert!( + actions.iter().all(|action| action.failures == 0), + "read failures occurred; see DEGRADED rows above" + ); +} diff --git a/crates/moon-core/src/db/tuner/mod.rs b/crates/moon-core/src/db/tuner/mod.rs index cde581c78..a88790ee5 100644 --- a/crates/moon-core/src/db/tuner/mod.rs +++ b/crates/moon-core/src/db/tuner/mod.rs @@ -434,6 +434,57 @@ pub(super) fn tuner_source_usdt_on(conn: &Connection, q: &Query) -> ReadResult ReadResult<(Query, String, DealOverlaySource)> { + let mut q = q.clone(); + q.floor_all_history(); + if q.metric != crate::db::ProfitMetric::Quote || !q.prefer_usdt { + let (q, src) = tuner_source_on(conn, &q)?; + return Ok((q, src, DealOverlaySource::Deferred)); + } + let (active, usdt) = crate::db::analytics::projection_with_usdt_on(conn, &q)?; + let src = + crate::db::analytics::unified_from_mode(conn, &q, active)?.ok_or(ReadFail::NotReady)?; + let usdt_src = match usdt { + Some(projection) if projection == active => Some(src.clone()), + Some(projection) => Some( + crate::db::analytics::unified_from_mode(conn, &q, projection)? + .ok_or(ReadFail::NotReady)?, + ), + None => None, + }; + Ok((q, src, DealOverlaySource::Resolved(usdt_src))) +} + +/// Snapshot-local overlay decision, deferred until active rows have been read when necessary. +enum DealOverlaySource { + /// The shared money preflight already resolved the overlay, including unavailable money. + Resolved(Option), + /// Active and overlay metrics require independent preflights. + Deferred, +} + +impl DealOverlaySource { + /// Resolve the overlay without repeating an already completed scope preflight. + fn resolve(self, conn: &Connection, q: &Query) -> ReadResult> { + match self { + Self::Resolved(source) => Ok(source), + Self::Deferred => tuner_source_usdt_on(conn, q), + } + } +} + /// Open a reader and materialize tuner rows in the same snapshot as quote preflight. /// /// The snapshot ends when `read` returns, before the caller performs CPU-heavy optimization. diff --git a/crates/moon-core/src/db/tuner/tests.rs b/crates/moon-core/src/db/tuner/tests.rs index 150765839..9c0a439f2 100644 --- a/crates/moon-core/src/db/tuner/tests.rs +++ b/crates/moon-core/src/db/tuner/tests.rs @@ -6,6 +6,95 @@ use rusqlite::trace::{TraceEvent, TraceEventCodes}; use super::*; +/// Sharing a money preflight must preserve both sources, split errors and measured-clock rows. +/// Reusing the active Percent source as USDT would change the second projection in this matrix. +#[test] +fn deal_sources_match_independent_preflights() { + use crate::db::test_support::{build_replica, remove_db, temp_db}; + use crate::db::valuation::ValuationMode; + let path = temp_db("deal-source-parity"); + let conn = build_replica(&path, &[(10_000, 3.0, "XALPHA"), (11_000, -2.0, "XALPHA")]); + conn.execute_batch( + "ALTER TABLE orders_rep ADD COLUMN channelname TEXT; + ALTER TABLE orders_rep ADD COLUMN signaltype TEXT; + UPDATE orders_rep SET core_name='CORE_A'; + UPDATE orders_rep SET channelname='LIQUIDATION', signaltype='UNIT_01', strategyid=0 WHERE newrecid=2; + ATTACH ':memory:' AS strat; + CREATE TABLE strat.strategies(core_uid INTEGER, strategy_id INTEGER, name TEXT, + deleted INTEGER, checked INTEGER); + INSERT INTO strat.strategies VALUES(1,7,'UNIT_01',1,0);", + ) + .expect("synthetic strategy fixture"); + let axis = crate::db::ReportAxis::from_measured( + std::collections::HashMap::from([( + 1, + vec![crate::db::OffsetSegment { + from_utc: 0, + offset_secs: 3600, + }], + )]), + chrono_tz::UTC, + ); + for split in [false, true] { + conn.execute( + "UPDATE orders_rep SET basecurrency=?1 WHERE newrecid=2", + [if split { 8 } else { 1 }], + ) + .expect("quote scope"); + for valuation in [ValuationMode::Historical, ValuationMode::Current] { + for metric in [ + crate::db::ProfitMetric::Quote, + crate::db::ProfitMetric::Percent, + ] { + for from in [-1, 6000] { + let query = Query { + axis: axis.clone(), + from, + to: 20_000, + prefer_usdt: true, + valuation, + metric, + ..Default::default() + }; + super::super::with_read_snapshot(&conn, |snapshot| { + let old = tuner_source_on(snapshot, &query).and_then(|(q, source)| { + Ok((q.clone(), source, tuner_source_usdt_on(snapshot, &q)?)) + }); + let new = + tuner_deal_sources_on(snapshot, &query).and_then(|(q, src, usdt)| { + let usdt = usdt.resolve(snapshot, &q)?; + Ok((q, src, usdt)) + }); + match (old, new) { + (Ok((old_q, old, old_usdt)), Ok((new_q, new, new_usdt))) => { + assert_eq!((old_q.from, old_q.to), (new_q.from, new_q.to)); + assert_eq!(old, new, "active projection"); + assert_eq!(old_usdt, new_usdt, "USDT overlay projection"); + let count: i64 = snapshot + .query_row( + &format!("SELECT COUNT(*) FROM {new}"), + rusqlite::params![new_q.from, new_q.to], + |r| r.get(0), + ) + .expect("materialize source"); + assert_eq!(count, 2); + } + (Err(old), Err(new)) => { + assert_eq!(format!("{old:?}"), format!("{new:?}")) + } + pair => panic!("preflight parity: {pair:?}"), + } + Ok(()) + }) + .expect("snapshot parity"); + } + } + } + } + drop(conn); + remove_db(&path); +} + /// Serializes the process-global function-pointer trace sink used by rusqlite. static TRACE_GUARD: Mutex<()> = Mutex::new(()); /// SQL statements executed while a tuner test connection has tracing enabled. diff --git a/crates/moon-core/src/db/tuner/ticks/deals.rs b/crates/moon-core/src/db/tuner/ticks/deals.rs index d93ac70a1..c947ff323 100644 --- a/crates/moon-core/src/db/tuner/ticks/deals.rs +++ b/crates/moon-core/src/db/tuner/ticks/deals.rs @@ -1,6 +1,6 @@ //! The axis' report rows: every closed trade of the tuner scope, as [`Deal`]s. //! -//! Read through the same unified source the other axes scan (`read_tuner_rows`), so the +//! Read through `tuner_deal_sources_on` inside one `with_read_snapshot`, so the //! "Fact" column and the tape replay describe the SAME trades — period, cores, strategies, //! emulator and side filters included. Three kinds of row are counted rather than dropped //! silently, and the caption and the load log print the counts: a row without a millisecond @@ -90,11 +90,14 @@ pub fn read_deals(q: &Query) -> ReadResult { // The scan on the tuner's own source (the metric decides `pnl`), then the USDT money of // every row off the USDT source in the same snapshot — the table's profit column must not // change unit with the scope's quote, and the scan's `profitbtc` would. - let mut read = crate::db::tuner::read_tuner_rows(q, |conn, q, src| { - let mut read = read_on(conn, q, src)?; - overlay_hook_detect(conn, q, &mut read.deals)?; - match crate::db::tuner::tuner_source_usdt_on(conn, q)? { - Some(usdt_src) => overlay_usdt_profit(conn, q, &usdt_src, &mut read.deals)?, + let conn = crate::db::open_reader()?; + let mut read = crate::db::with_read_snapshot(&conn, |conn| { + let (q, src, usdt_src) = crate::db::tuner::tuner_deal_sources_on(conn, q)?; + let mut read = read_on(conn, &q, &src)?; + overlay_hook_detect(conn, &q, &mut read.deals)?; + let usdt_src = usdt_src.resolve(conn, &q)?; + match usdt_src { + Some(usdt_src) => overlay_usdt_profit(conn, &q, &usdt_src, &mut read.deals)?, None => log::info!( target: crate::diagnostics::TICKS_AXIS_TARGET, "[x] ticks deals: the scope's money cannot be valued in USDT, the profit column stays empty" diff --git a/crates/moon-core/src/db/valuation/coverage.rs b/crates/moon-core/src/db/valuation/coverage.rs index f0d8ce46b..58b9f13cf 100644 --- a/crates/moon-core/src/db/valuation/coverage.rs +++ b/crates/moon-core/src/db/valuation/coverage.rs @@ -198,6 +198,64 @@ pub(in crate::db) struct SourcePredicates { pub(in crate::db) fn source_predicates( alias: &str, columns: &std::collections::HashSet, +) -> SourcePredicates { + source_predicates_with_inputs(alias, columns, &CoverageInputs::report(alias, columns)) +} + +/// Original cells retained beside normalized amounts for storage guards and cache identity. +pub(in crate::db) const RAW_COVERAGE_COLUMNS: &[&str] = &[ + "closedate", + "profitbtc", + "spentbtc", + "core_uid", + "newrecid", + "db_id", +]; + +/// Expressions whose storage checks remain on the original cells while amounts are normalized. +pub(in crate::db) struct CoverageInputs { + /// Trusted effective quote ordinal, including denomination corrections. + pub quote: String, + /// Settled profit, before conversion. + pub profit: String, + /// Settled spend, before conversion and numeric storage validation. + pub spent: String, +} + +impl CoverageInputs { + /// Name the normalization-stage columns, optionally qualified by a row alias. + pub(in crate::db) fn normalized(alias: &str) -> Self { + let prefix = if alias.is_empty() { + String::new() + } else { + format!("{alias}.") + }; + Self { + quote: format!("{prefix}normalized_quote"), + profit: format!("{prefix}settled_profit"), + spent: format!("{prefix}settled_spent"), + } + } + + /// Build the original report expressions for callers without a normalization stage. + pub(in crate::db) fn report(alias: &str, columns: &std::collections::HashSet) -> Self { + Self { + quote: crate::db::quote::effective_ordinal_expr(alias, columns), + profit: crate::db::quote::settled_amount_expr(alias, columns, "profitbtc"), + spent: if columns.contains("spentbtc") { + crate::db::quote::settled_amount_expr(alias, columns, "spentbtc") + } else { + "NULL".to_string() + }, + } + } +} + +/// Build storage guards against raw cells and values against explicit normalized expressions. +pub(in crate::db) fn source_predicates_with_inputs( + alias: &str, + columns: &std::collections::HashSet, + inputs: &CoverageInputs, ) -> SourcePredicates { // SQLite resolves every column reference at prepare time, unreachable arms included, so an // absent column collapses to a constant rather than being named. @@ -205,10 +263,7 @@ pub(in crate::db) fn source_predicates( quote_known: if columns.contains("basecurrency") { // The effective expression already rejects every non-integer storage class, so the // range check alone decides whether the identity is one this build knows. - format!( - "({quote}) BETWEEN 0 AND 20", - quote = crate::db::quote::effective_ordinal_expr(alias, columns) - ) + format!("({quote}) BETWEEN 0 AND 20", quote = inputs.quote) } else { "0".to_string() }, @@ -223,7 +278,7 @@ pub(in crate::db) fn source_predicates( spent_value: if columns.contains("spentbtc") { format!( "CASE WHEN typeof({alias}.spentbtc) IN ('integer','real') THEN {settled} END", - settled = crate::db::quote::settled_amount_expr(alias, columns, "spentbtc") + settled = inputs.spent ) } else { "NULL".to_string() @@ -244,6 +299,21 @@ pub(crate) fn coverage_sql( alias: &str, columns: &std::collections::HashSet, source: TradeSource, +) -> CoverageSql { + coverage_sql_with_inputs( + alias, + columns, + source, + &CoverageInputs::report(alias, columns), + ) +} + +/// Build historical coverage using already normalized quote and amount expressions. +fn coverage_sql_with_inputs( + alias: &str, + columns: &std::collections::HashSet, + source: TradeSource, + inputs: &CoverageInputs, ) -> CoverageSql { let id_column = match source { TradeSource::Typed if columns.contains("newrecid") => Some("newrecid"), @@ -256,12 +326,12 @@ pub(crate) fn coverage_sql( // Every quote reference below goes through this ONE expression: the cache is keyed by quote // ordinal, so a join that matched the raw column while the projection valued the effective one // would look up the rate of a currency the row is not denominated in. - let quote = crate::db::quote::effective_ordinal_expr(alias, columns); + let quote = &inputs.quote; let SourcePredicates { quote_known, numeric_profit, spent_value, - } = source_predicates(alias, columns); + } = source_predicates_with_inputs(alias, columns, inputs); let date_valid = if has_closedate { format!("typeof({alias}.closedate)='integer' AND {alias}.closedate>0") } else { @@ -280,7 +350,7 @@ pub(crate) fn coverage_sql( AND v.profit_quote={settled_profit} AND v.spent_quote IS {spent_value}", source_kind = source.code(), algorithm_version = ALGORITHM_VERSION, - settled_profit = crate::db::quote::settled_amount_expr(alias, columns, "profitbtc"), + settled_profit = inputs.profit, ) } else { "0".to_string() @@ -304,7 +374,7 @@ pub(crate) fn coverage_sql( format!( "CASE WHEN {identity} THEN {settled} WHEN {prepared} THEN v.profit_usdt END", - settled = crate::db::quote::settled_amount_expr(alias, columns, "profitbtc") + settled = inputs.profit ) } else { "NULL".to_string() @@ -413,6 +483,28 @@ pub(in crate::db) fn projection( } } +/// Build either coverage mode over explicit normalization-stage columns. +pub(in crate::db) fn projection_with_inputs( + mode: ValuationMode, + attached: bool, + alias: &str, + columns: &std::collections::HashSet, + source: TradeSource, + inputs: &CoverageInputs, +) -> Option { + match mode { + ValuationMode::Historical => { + attached.then(|| coverage_sql_with_inputs(alias, columns, source, inputs)) + } + ValuationMode::Current => { + let (rates, now_ms) = current::current_rates_at(); + Some(current::current_rate_sql_with_inputs( + alias, columns, &rates, now_ms, inputs, + )) + } + } +} + /// Report columns a trade must carry before it can be valued at all. /// /// Shared with the Report window's synthetic-column table, so the columns that OFFER a conversion diff --git a/crates/moon-core/src/db/valuation/current.rs b/crates/moon-core/src/db/valuation/current.rs index b0355ce35..e456f38bd 100644 --- a/crates/moon-core/src/db/valuation/current.rs +++ b/crates/moon-core/src/db/valuation/current.rs @@ -402,6 +402,23 @@ pub(crate) fn current_rate_sql( columns: &std::collections::HashSet, rates: &CurrentRates, now_ms: i64, +) -> CoverageSql { + current_rate_sql_with_inputs( + alias, + columns, + rates, + now_ms, + &super::CoverageInputs::report(alias, columns), + ) +} + +/// Build current coverage over explicit normalized values, retaining raw-cell storage checks. +pub(super) fn current_rate_sql_with_inputs( + alias: &str, + columns: &std::collections::HashSet, + rates: &CurrentRates, + now_ms: i64, + inputs: &super::CoverageInputs, ) -> CoverageSql { let has_quote = columns.contains("basecurrency"); let has_profit = columns.contains("profitbtc"); @@ -410,7 +427,7 @@ pub(crate) fn current_rate_sql( quote_known, numeric_profit, spent_value, - } = super::source_predicates(alias, columns); + } = super::source_predicates_with_inputs(alias, columns, inputs); // Without a quote column there is nothing to switch on, and naming it in an unreachable arm // would still fail at prepare time — so the arms are not even built. @@ -435,7 +452,7 @@ pub(crate) fn current_rate_sql( } // Switched on the EFFECTIVE ordinal, like every other quote reference: a COIN-M row is // denominated in BTC, so it must pick the BTC arm rather than the identity one. - let quote = super::super::quote::effective_ordinal_expr(alias, columns); + let quote = &inputs.quote; ( format!("(CASE ({quote}){rate_arms} END)"), format!("(CASE ({quote}){source_arms} END)"), @@ -454,7 +471,7 @@ pub(crate) fn current_rate_sql( "({eligible} AND {numeric_profit} AND {rate_case} IS NULL AND ({quote}) IN ({}))", unroutable.join(","), - quote = super::super::quote::effective_ordinal_expr(alias, columns) + quote = inputs.quote ) } else { "0".to_string() @@ -462,7 +479,7 @@ pub(crate) fn current_rate_sql( let profit_usdt = if has_profit { format!( "CASE WHEN {valued} THEN ({settled}) * {rate_case} END", - settled = super::super::quote::settled_amount_expr(alias, columns, "profitbtc") + settled = inputs.profit ) } else { "NULL".to_string()