perf(analytics): read deals, tuner lists and the calendar faster at large report volumes

crates/moon-core/src/db/analytics/groups.rs

@@ -361,6 +361,25 @@ pub(super) fn groups(
     by_strategy: bool,
 ) -> ReadResult<Vec<GroupStat>> {
     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,

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::<f64>(), 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 {

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<ScopeDecision> {
+    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<QuoteBreakdown>,
+) -> ReadResult<ScopeDecision> {
     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,15 +491,32 @@ 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<ProjectionMode> {
+    match decision {
         ScopeDecision::Comparable { unit, projection }
             if unit == ProfitUnit::Quote(crate::db::QuoteCurrency::usdt()) =>
         {
             Some(projection)
         }
         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<ProjectionMode>)> {
+    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.

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,
     };

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<QuoteBreakdown> {
+    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<QuoteBreakdown> {
+    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<QuoteBreakdown> {
     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<QuoteBreakdown> {
     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::<Vec<_>>()
+            .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::<Vec<_>>()