diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/InventoryTurnoverSql.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/InventoryTurnoverSql.java
index 665388449..cc7fdb683 100644
--- a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/InventoryTurnoverSql.java
+++ b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/InventoryTurnoverSql.java
@@ -38,39 +38,38 @@ public final class InventoryTurnoverSql {
             +"CAST(#{p.days["+i+"].coverage} AS SIGNED) coverage,CAST(#{p.days["+i+"].sampleRun,jdbcType=BIGINT} AS SIGNED) sample_run");
         return String.join(" UNION ALL ",parts);
     }
+    private static String sampleRuns(Input p) {
+        // The frozen context already resolved these versions. Bind them directly instead
+        // of repeatedly expanding the 30-day calendar as a membership subquery.
+        var runs=new ArrayList<String>();
+        for(int i=0;i<p.days().size();i++) if(p.days().get(i).sampleRun()!=null)
+            runs.add("#{p.days["+i+"].sampleRun,jdbcType=BIGINT}");
+        return runs.isEmpty()?"NULL":String.join(",",runs);
+    }
+    private static String salesDays() {
+        return "SELECT customer_code,sku_code,business_date,SUM(outbound_quantity) quantity,MIN(CASE WHEN quantity_unit='PCS' AND outbound_quantity>=0 THEN 1 ELSE 0 END) unit_valid FROM ll_selfop_sku_daily_sales WHERE "
+            +scope("")+" AND sales_run_id=#{p.salesRun} AND business_date>=CAST(#{p.from} AS DATE) AND business_date<CAST(#{p.until} AS DATE) GROUP BY customer_code,sku_code,business_date";
+    }
     private static String cte(Input p) {
         String stocks="SELECT snapshot_run_id,customer_code,sku_code,"+amount("physical_quantity")+" physical,"
             +amount("frozen_quantity")+" frozen,MAX(CASE WHEN product_quality=0 AND "+valid("available_quantity")+" THEN available_quantity END) available,"
             +"CASE WHEN MIN(CASE WHEN product_quality NOT IN (0,1) OR product_quality IS NULL THEN 0 WHEN product_quality=1 AND NOT ("+valid("physical_quantity")+") THEN 0 ELSE 1 END)=1 "
             +"THEN SUM(CASE WHEN product_quality=1 THEN physical_quantity ELSE 0 END) END defective "
-            +"FROM ll_selfop_inventory_snapshot WHERE "+scope("")+" AND (snapshot_run_id=#{p.inventoryRun} OR snapshot_run_id IN (SELECT sample_run FROM calendar)) "
+            +"FROM ll_selfop_inventory_snapshot WHERE "+scope("")+" AND (snapshot_run_id=#{p.inventoryRun} OR snapshot_run_id IN ("+sampleRuns(p)+")) "
             +"GROUP BY snapshot_run_id,customer_code,sku_code";
-        String profiles="SELECT * FROM ll_selfop_sku_profile WHERE "+scope("")+" AND (profile_run_id IN (#{p.salesRun},#{p.inventoryRun}) OR profile_run_id IN (SELECT sample_run FROM calendar))";
+        String profiles="SELECT * FROM ll_selfop_sku_profile WHERE "+scope("")+" AND (profile_run_id IN (#{p.salesRun},#{p.inventoryRun}) OR profile_run_id IN ("+sampleRuns(p)+"))";
         StringBuilder sql=new StringBuilder("WITH calendar AS (").append(calendar(p)).append("), fixed_profiles AS (").append(profiles)
-            .append("), stocks AS (").append(stocks).append("), sales_days AS (SELECT customer_code,sku_code,business_date,SUM(outbound_quantity) quantity,MIN(CASE WHEN quantity_unit='PCS' AND outbound_quantity>=0 THEN 1 ELSE 0 END) unit_valid FROM ll_selfop_sku_daily_sales WHERE ")
-            .append(scope("")).append(" AND sales_run_id=#{p.salesRun} AND business_date>=CAST(#{p.from} AS DATE) AND business_date<CAST(#{p.until} AS DATE) GROUP BY customer_code,sku_code,business_date), candidates AS (")
+            .append("), stocks AS (").append(stocks).append("), sales_days AS (").append(salesDays()).append("), candidates AS (")
             .append("SELECT customer_code,sku_code FROM fixed_profiles UNION SELECT customer_code,sku_code FROM stocks UNION SELECT customer_code,sku_code FROM sales_days), profiles AS (")
             .append("SELECT c.customer_code,c.sku_code,COALESCE(MAX(CASE WHEN f.profile_run_id=#{p.inventoryRun} THEN f.product_name END),MAX(f.product_name)) product_name,")
             .append("COALESCE(MAX(CASE WHEN f.profile_run_id=#{p.salesRun} THEN CASE WHEN f.quantity_unit='PCS' THEN 1 ELSE 0 END END),MAX(CASE WHEN f.profile_run_id=#{p.inventoryRun} THEN CASE WHEN f.quantity_unit='PCS' THEN 1 ELSE 0 END END),MIN(CASE WHEN f.quantity_unit='PCS' THEN 1 ELSE 0 END)) unit_valid,")
             .append("MIN(CASE WHEN f.evidence_run_id IS NOT NULL AND f.evidence_type IN ('VALID_OUTBOUND','INVENTORY_OBSERVATION') THEN f.earliest_evidence_at_utc END) evidence ")
-            .append("FROM candidates c LEFT JOIN fixed_profiles f ON f.customer_code=c.customer_code AND f.sku_code=c.sku_code GROUP BY c.customer_code,c.sku_code), daily_values AS (")
-            .append("SELECT c.customer_code,c.sku_code,d.business_date,d.coverage,c.unit_valid,c.evidence,sd.quantity,sd.unit_valid sales_unit,")
-            .append("CASE WHEN ip.quantity_unit='PCS' THEN hs.physical END sample_physical ")
-            .append("FROM profiles c CROSS JOIN calendar d LEFT JOIN sales_days sd ON sd.customer_code=c.customer_code AND sd.sku_code=c.sku_code AND sd.business_date=d.business_date ")
-            .append("LEFT JOIN stocks hs ON hs.snapshot_run_id=d.sample_run AND hs.customer_code=c.customer_code AND hs.sku_code=c.sku_code ")
-            .append("LEFT JOIN fixed_profiles ip ON ip.profile_run_id=d.sample_run AND ip.customer_code=c.customer_code AND ip.sku_code=c.sku_code), day_result AS (")
-            .append("SELECT v.*,CASE WHEN #{p.salesKnown}=1 AND unit_valid=1 AND COALESCE(sales_unit,1)=1 THEN CASE WHEN quantity IS NOT NULL THEN quantity WHEN coverage>0 AND evidence<=CONVERT_TZ(CAST(business_date AS DATETIME),'+08:00','+00:00') THEN 0 END END observed,")
-            .append("CASE WHEN coverage=1 AND unit_valid=1 AND COALESCE(sales_unit,1)=1 AND (quantity IS NOT NULL OR evidence<=CONVERT_TZ(CAST(business_date AS DATETIME),'+08:00','+00:00')) THEN 1 ELSE 0 END complete,")
-            .append("CASE WHEN #{p.salesKnown}=1 AND coverage=1 AND COALESCE(sales_unit,1)=1 THEN 1 ELSE 0 END owner_complete,")
-            .append("CASE WHEN #{p.salesKnown}=1 AND COALESCE(sales_unit,1)=1 THEN CASE WHEN quantity IS NOT NULL THEN quantity WHEN coverage>0 THEN 0 END END owner_observed FROM daily_values v), sku_windows AS (")
-            .append("SELECT customer_code,sku_code").append(windowAggregates(false)).append(",COUNT(CASE WHEN business_date>=CAST(#{p.baseFrom} AS DATE) THEN sample_physical END) sample_count,")
-            .append("CASE WHEN COUNT(CASE WHEN business_date>=CAST(#{p.baseFrom} AS DATE) THEN sample_physical END)=#{p.query.days} THEN SUM(CASE WHEN business_date>=CAST(#{p.baseFrom} AS DATE) THEN sample_physical END)/#{p.query.days} END average_physical,")
-            .append("GROUP_CONCAT(CASE WHEN business_date>=CAST(#{p.baseFrom} AS DATE) AND sample_physical IS NULL THEN CAST(business_date AS CHAR) END ORDER BY business_date SEPARATOR ',') missing_dates ")
-            .append("FROM day_result GROUP BY customer_code,sku_code), last_sales AS (SELECT customer_code,sku_code,MAX(business_date) last_outbound FROM ll_selfop_sku_daily_sales WHERE ")
+            .append("FROM candidates c LEFT JOIN fixed_profiles f ON f.customer_code=c.customer_code AND f.sku_code=c.sku_code GROUP BY c.customer_code,c.sku_code)")
+            .append(skuWindows()).append(", last_sales AS (SELECT customer_code,sku_code,MAX(business_date) last_outbound FROM ll_selfop_sku_daily_sales WHERE ")
             .append(scope("")).append(" AND sales_run_id=#{p.salesRun} AND business_date<CAST(#{p.until} AS DATE) AND outbound_quantity>0 AND quantity_unit='PCS' GROUP BY customer_code,sku_code)");
         sql.append(", sku_base AS (SELECT c.customer_code,c.sku_code,c.product_name,c.evidence,c.unit_valid");
         for(int n:List.of(7,15,30)) for(String suffix:List.of("outbound","observed","complete","partial")) sql.append(",w.w").append(n).append('_').append(suffix);
-        sql.append(",w.sample_count,w.average_physical,w.missing_dates,ls.last_outbound");
+        sql.append(",w.sample_count,w.average_physical,w.sample_mask,ls.last_outbound");
         for(String field:List.of("physical","available","frozen","defective")) sql.append(",CASE WHEN #{p.stockFresh}=1 AND ip.quantity_unit='PCS' THEN st.").append(field).append(" END ").append(field);
         sql.append(" FROM profiles c JOIN sku_windows w ON w.customer_code=c.customer_code AND w.sku_code=c.sku_code LEFT JOIN last_sales ls ON ls.customer_code=c.customer_code AND ls.sku_code=c.sku_code LEFT JOIN stocks st ON st.snapshot_run_id=#{p.inventoryRun} AND st.customer_code=c.customer_code AND st.sku_code=c.sku_code ")
             .append("LEFT JOIN fixed_profiles ip ON ip.profile_run_id=#{p.inventoryRun} AND ip.customer_code=c.customer_code AND ip.sku_code=c.sku_code), sku_calculated AS (SELECT b.*,")
@@ -82,26 +81,95 @@ public final class InventoryTurnoverSql {
             .append(p.query.days()).append("_outbound=0 AND unassessable=0 THEN 0 END slow FROM sku_calculated b) ");
         return sql.toString();
     }
-    private static String windowAggregates(boolean owner) {
+    private static String skuWindows() {
+        // One shared suffix table qualifies absent sales dates. Millisecond evidence
+        // after a midnight selects the next midnight, never the truncated same date.
+        var sql=new StringBuilder(", evidence_boundaries AS (SELECT start_utc,LAG(start_utc) OVER (ORDER BY start_utc) previous_start");
+        for(int n:List.of(7,15,30)) for(String kind:List.of("complete","known","partial")) {
+            String coverage=switch(kind){case "complete"->"coverage=1";case "known"->"coverage>0";default->"coverage=2";};
+            sql.append(",SUM(CASE WHEN business_date>=CAST(#{p.from").append(n).append("} AS DATE) AND ").append(coverage)
+                .append(" THEN 1 ELSE 0 END) OVER (ORDER BY start_utc ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING) w").append(n).append('_').append(kind);
+        }
+        sql.append(" FROM calendar), sku_zero_counts AS (SELECT c.customer_code,c.sku_code");
+        for(int n:List.of(7,15,30)) for(String kind:List.of("complete","known","partial"))
+            sql.append(",CASE WHEN c.unit_valid=1").append("complete".equals(kind)?"":" AND #{p.salesKnown}=1")
+                .append(" THEN COALESCE(b.w").append(n).append('_').append(kind).append(",0) ELSE 0 END w").append(n).append('_').append(kind);
+        sql.append(" FROM profiles c LEFT JOIN evidence_boundaries b ON c.evidence<=b.start_utc AND (b.previous_start IS NULL OR c.evidence>b.previous_start)), actual_sku_days AS (")
+            .append("SELECT s.customer_code,s.sku_code,s.business_date,CASE WHEN #{p.salesKnown}=1 AND c.unit_valid=1 AND s.unit_valid=1 THEN s.quantity END observed,")
+            .append("CASE WHEN d.coverage=1 AND c.unit_valid=1 AND s.unit_valid=1 THEN 1 ELSE 0 END complete,")
+            .append("CASE WHEN d.coverage=1 AND c.unit_valid=1 AND c.evidence<=d.start_utc THEN 1 ELSE 0 END zero_complete,")
+            .append("CASE WHEN #{p.salesKnown}=1 AND c.unit_valid=1 AND d.coverage>0 AND c.evidence<=d.start_utc THEN 1 ELSE 0 END zero_known,")
+            .append("CASE WHEN #{p.salesKnown}=1 AND c.unit_valid=1 AND d.coverage=2 AND c.evidence<=d.start_utc THEN 1 ELSE 0 END zero_partial ")
+            .append("FROM sales_days s JOIN profiles c ON c.customer_code=s.customer_code AND c.sku_code=s.sku_code JOIN calendar d ON d.business_date=s.business_date), sku_actual_counts AS (SELECT customer_code,sku_code");
+        for(int n:List.of(7,15,30)) {
+            String range="business_date>=CAST(#{p.from"+n+"} AS DATE)";
+            sql.append(",SUM(CASE WHEN ").append(range).append(" THEN COALESCE(observed,0) ELSE 0 END) w").append(n).append("_sum")
+                .append(",SUM(CASE WHEN ").append(range).append(" THEN complete-zero_complete ELSE 0 END) w").append(n).append("_complete")
+                .append(",SUM(CASE WHEN ").append(range).append(" THEN (CASE WHEN observed IS NOT NULL THEN 1 ELSE 0 END)-zero_known ELSE 0 END) w").append(n).append("_known")
+                .append(",SUM(CASE WHEN ").append(range).append(" THEN (CASE WHEN observed IS NOT NULL AND complete=0 THEN 1 ELSE 0 END)-zero_partial ELSE 0 END) w").append(n).append("_partial");
+        }
+        // Read only actual qualified samples at the frozen date/run. Missing candidates
+        // remain in sku_zero_counts; absence never creates a zero-valued sample.
+        sql.append(" FROM actual_sku_days GROUP BY customer_code,sku_code), sku_sample_days AS (SELECT st.customer_code,st.sku_code,d.business_date,")
+            .append("st.physical sample_physical FROM calendar d JOIN stocks st ON st.snapshot_run_id=d.sample_run ")
+            .append("JOIN fixed_profiles ip ON ip.profile_run_id=d.sample_run AND ip.customer_code=st.customer_code AND ip.sku_code=st.sku_code ")
+            .append("WHERE ip.quantity_unit='PCS' AND st.physical IS NOT NULL), sku_sample_counts AS (SELECT customer_code,sku_code,")
+            .append("COUNT(CASE WHEN business_date>=CAST(#{p.baseFrom} AS DATE) THEN sample_physical END) sample_count,")
+            .append("CASE WHEN COUNT(CASE WHEN business_date>=CAST(#{p.baseFrom} AS DATE) THEN sample_physical END)=#{p.query.days} THEN SUM(CASE WHEN business_date>=CAST(#{p.baseFrom} AS DATE) THEN sample_physical END)/#{p.query.days} END average_physical,")
+            .append(sampleMask()).append(" sample_mask FROM sku_sample_days GROUP BY customer_code,sku_code), sku_window_counts AS (SELECT z.customer_code,z.sku_code,COALESCE(s.sample_count,0) sample_count,s.average_physical,COALESCE(s.sample_mask,0) sample_mask");
+        for(int n:List.of(7,15,30)) sql.append(",CASE WHEN z.w").append(n).append("_known+COALESCE(a.w").append(n).append("_known,0)>0 THEN COALESCE(a.w").append(n).append("_sum,0) END w").append(n).append("_observed")
+            .append(",z.w").append(n).append("_complete+COALESCE(a.w").append(n).append("_complete,0) w").append(n).append("_complete")
+            .append(",z.w").append(n).append("_partial+COALESCE(a.w").append(n).append("_partial,0) w").append(n).append("_partial");
+        return sql.append(" FROM sku_zero_counts z LEFT JOIN sku_actual_counts a ON a.customer_code=z.customer_code AND a.sku_code=z.sku_code ")
+            .append("LEFT JOIN sku_sample_counts s ON s.customer_code=z.customer_code AND s.sku_code=z.sku_code), sku_windows AS (SELECT w.*")
+            .append(formalWindows()).append(" FROM sku_window_counts w)").toString();
+    }
+    private static String sampleMask() {
+        // Exactly 30 bits, all exactly representable even by POWER's numeric result.
+        return "SUM(CASE WHEN sample_physical IS NOT NULL THEN CAST(POWER(2,TIMESTAMPDIFF(DAY,CAST(#{p.from} AS DATE),business_date)) AS SIGNED) ELSE 0 END)";
+    }
+    private static String pageMissingDates() {
+        // Only the selected page expands dates. Numeric qualification and filtering
+        // already used every candidate and every frozen sample before pagination.
+        return ", page_missing AS (SELECT r.*, (SELECT GROUP_CONCAT(CAST(business_date AS CHAR) ORDER BY business_date SEPARATOR ',') FROM calendar d "
+            +"WHERE d.business_date>=CAST(#{p.baseFrom} AS DATE) AND MOD(FLOOR(r.sample_mask/POWER(2,TIMESTAMPDIFF(DAY,CAST(#{p.from} AS DATE),d.business_date))),2)=0) missing_dates FROM page_result r)";
+    }
+    private static String windowAggregates() {
         StringBuilder sql=new StringBuilder();
         for(int n:List.of(7,15,30)) {
             String range="business_date>=CAST(#{p.from"+n+"} AS DATE)";
-            sql.append(",CASE WHEN SUM(CASE WHEN ").append(range).append(" THEN complete ELSE 0 END)=").append(n)
-                .append(" THEN SUM(CASE WHEN ").append(range).append(" THEN COALESCE(observed,0) END) END w").append(n).append("_outbound")
-                .append(",SUM(CASE WHEN ").append(range).append(" THEN observed END) w").append(n).append("_observed")
+            sql.append(",SUM(CASE WHEN ").append(range).append(" THEN observed END) w").append(n).append("_observed")
                 .append(",SUM(CASE WHEN ").append(range).append(" THEN complete ELSE 0 END) w").append(n).append("_complete")
                 .append(",SUM(CASE WHEN ").append(range).append(" AND complete=0 AND observed IS NOT NULL THEN 1 ELSE 0 END) w").append(n).append("_partial");
         }
         return sql.toString();
     }
+    private static String formalWindows() {
+        // Derive formal values once per SKU/owner, reusing the same complete count and
+        // observed sum. A complete all-null observation sum retains the prior zero;
+        // partial/unknown windows remain null and never shorten the denominator.
+        var sql=new StringBuilder();
+        for(int n:List.of(7,15,30)) sql.append(",CASE WHEN w.w").append(n).append("_complete=").append(n)
+            .append(" THEN COALESCE(w.w").append(n).append("_observed,0) END w").append(n).append("_outbound");
+        return sql.toString();
+    }
     private static String ownerCte(Input p) {
-        return ", owner_days AS (SELECT customer_code,business_date,MIN(owner_complete) complete,"
-            +"SUM(owner_observed) observed,"
-            +"CASE WHEN COUNT(sample_physical)=COUNT(*) THEN SUM(sample_physical) END sample_physical FROM day_result GROUP BY customer_code,business_date), owner_windows AS (SELECT customer_code"
-            +windowAggregates(true)+",COUNT(CASE WHEN business_date>=CAST(#{p.baseFrom} AS DATE) THEN sample_physical END) sample_count,"
+        // Sales units apply to actual daily records, not profile units/evidence.
+        // The full candidate count distinguishes covered absent rows (known zero)
+        // from a day where every candidate has only an invalid actual record.
+        return ", owner_candidates AS (SELECT customer_code,COUNT(*) candidate_count FROM candidates GROUP BY customer_code), owner_actual_days AS ("
+            +"SELECT customer_code,business_date,COUNT(*) actual_count,SUM(CASE WHEN unit_valid=1 THEN 1 ELSE 0 END) valid_count,"
+            +"SUM(CASE WHEN unit_valid=1 THEN quantity END) observed FROM sales_days GROUP BY customer_code,business_date), owner_sample_days AS ("
+            +"SELECT s.customer_code,s.business_date,SUM(s.sample_physical) sample_physical "
+            +"FROM sku_sample_days s JOIN owner_candidates c ON c.customer_code=s.customer_code GROUP BY s.customer_code,s.business_date HAVING COUNT(*)=MAX(c.candidate_count)), owner_days AS ("
+            +"SELECT c.customer_code,d.business_date,CASE WHEN #{p.salesKnown}=1 AND d.coverage=1 AND COALESCE(a.actual_count-a.valid_count,0)=0 THEN 1 ELSE 0 END complete,"
+            +"CASE WHEN #{p.salesKnown}=1 THEN CASE WHEN a.valid_count>0 THEN a.observed WHEN d.coverage>0 AND c.candidate_count>COALESCE(a.actual_count,0) THEN 0 END END observed,"
+            +"s.sample_physical FROM owner_candidates c CROSS JOIN calendar d LEFT JOIN owner_actual_days a ON a.customer_code=c.customer_code AND a.business_date=d.business_date "
+            +"LEFT JOIN owner_sample_days s ON s.customer_code=c.customer_code AND s.business_date=d.business_date), owner_window_counts AS (SELECT customer_code"
+            +windowAggregates()+",COUNT(CASE WHEN business_date>=CAST(#{p.baseFrom} AS DATE) THEN sample_physical END) sample_count,"
             +"CASE WHEN COUNT(CASE WHEN business_date>=CAST(#{p.baseFrom} AS DATE) THEN sample_physical END)=#{p.query.days} THEN SUM(CASE WHEN business_date>=CAST(#{p.baseFrom} AS DATE) THEN sample_physical END)/#{p.query.days} END average_physical,"
-            +"GROUP_CONCAT(CASE WHEN business_date>=CAST(#{p.baseFrom} AS DATE) AND sample_physical IS NULL THEN CAST(business_date AS CHAR) END ORDER BY business_date SEPARATOR ',') missing_dates FROM owner_days GROUP BY customer_code), owner_totals AS ("
-            +totals("customer_code",true)+" FROM sku_result GROUP BY customer_code), owner_result AS (SELECT t.*,w.sample_count,w.average_physical,w.missing_dates,"
+            +sampleMask()+" sample_mask FROM owner_days GROUP BY customer_code), owner_windows AS (SELECT w.*"+formalWindows()+" FROM owner_window_counts w), owner_totals AS ("
+            +totals("customer_code",true)+" FROM sku_result GROUP BY customer_code), owner_result AS (SELECT t.*,w.sample_count,w.average_physical,w.sample_mask,"
             +"CASE WHEN w.average_physical>0 THEN w.w"+p.query.days()+"_outbound/w.average_physical END turnover,"
             +String.join(",",List.of(7,15,30).stream().flatMap(n->List.of("outbound","observed","complete","partial").stream().map(s->"w.w"+n+"_"+s)).toList())
             +" FROM owner_totals t JOIN owner_windows w ON w.customer_code=t.customer_code) ";
@@ -140,7 +208,7 @@ public final class InventoryTurnoverSql {
     }
     public static String page(Map<String,Object> args) {
         Input p=(Input)args.get("p");boolean owner="OWNER".equals(p.query.group());
-        return cte(p)+(owner?ownerCte(p):"")+"SELECT * FROM "+(owner?"owner_result":"sku_result")+filter(p,owner)+order(p,owner)+" LIMIT #{p.query.pageSize} OFFSET #{p.query.offset}";
+        return cte(p)+(owner?ownerCte(p):"")+", page_result AS (SELECT * FROM "+(owner?"owner_result":"sku_result")+filter(p,owner)+order(p,owner)+" LIMIT #{p.query.pageSize} OFFSET #{p.query.offset})"+pageMissingDates()+" SELECT * FROM page_missing"+order(p,owner);
     }
     public static String count(Map<String,Object> args) {
         Input p=(Input)args.get("p");boolean owner="OWNER".equals(p.query.group());
@@ -155,10 +223,10 @@ public final class InventoryTurnoverSql {
             +"SELECT 'OWNER' kind,customer_code,physical,0 unknown_count FROM ranked_owners UNION ALL SELECT 'META' kind,NULL customer_code,NULL physical,COUNT(*) unknown_count FROM owner_result WHERE physical IS NULL ORDER BY kind DESC,physical DESC,customer_code";
     }
     public static String chartWindows(Map<String,Object> args) {
-        Input p=(Input)args.get("p");return cte(p)+", warehouse_days AS (SELECT d.business_date,"
+        Input p=(Input)args.get("p");return "WITH calendar AS ("+calendar(p)+"), sales_days AS ("+salesDays()+"), warehouse_days AS (SELECT d.business_date,"
             +"CASE WHEN #{p.salesKnown}=1 AND d.coverage=1 AND MIN(COALESCE(s.unit_valid,1))=1 THEN 1 ELSE 0 END complete,"
             +"CASE WHEN #{p.salesKnown}=1 THEN CASE WHEN COUNT(s.business_date)=0 AND d.coverage>0 THEN 0 ELSE SUM(CASE WHEN s.unit_valid=1 THEN s.quantity END) END END observed "
-            +"FROM calendar d LEFT JOIN sales_days s ON s.business_date=d.business_date GROUP BY d.business_date,d.coverage) SELECT COUNT(*) day_count"+windowAggregates(true)+" FROM warehouse_days";
+            +"FROM calendar d LEFT JOIN sales_days s ON s.business_date=d.business_date GROUP BY d.business_date,d.coverage), warehouse_window_counts AS (SELECT COUNT(*) day_count"+windowAggregates()+" FROM warehouse_days) SELECT w.*"+formalWindows()+" FROM warehouse_window_counts w";
     }
     private static String jsonFields(List<String> fields) {
         return "JSON_OBJECT("+String.join(",",fields.stream().map(f->"'"+f+"',"+f).toList())+")";
@@ -174,17 +242,21 @@ public final class InventoryTurnoverSql {
             for(String field:List.of("physical","available","frozen","defective"))for(String suffix:List.of("known","unknown"))fields.add(field+"_"+suffix);
             for(String risk:List.of("no_outbound","slow"))for(String suffix:List.of("count","physical","unknown"))fields.add(risk+"_"+suffix);
         } else fields.addAll(List.of("sku_code","product_name","coverage_days","no_outbound","slow"));
-        String sql=cte(p)+(owner||overview?ownerCte(p):"")
+        // The physical-stock chart needs owner totals, not owner sales/sample windows.
+        String sql=cte(p)+(owner?ownerCte(p):overview?", owner_totals AS ("+totals("customer_code",true)+" FROM sku_result GROUP BY customer_code) ":"")
             +", page_result AS (SELECT r.*,ROW_NUMBER() OVER ("+order(p,owner)+") ordinal FROM "+source+" r"+filter(p,owner)+order(p,owner)+" LIMIT #{p.query.pageSize} OFFSET #{p.query.offset})";
-        if(overview)sql+=", summary_result AS ("+totals(null,false)+" FROM sku_result), top_owners AS (SELECT customer_code,physical,ROW_NUMBER() OVER (ORDER BY physical DESC,customer_code) ordinal FROM owner_result WHERE physical IS NOT NULL ORDER BY physical DESC,customer_code LIMIT 10)";
-        sql+=" SELECT 'ROW' section,ordinal,"+jsonFields(fields)+" payload FROM page_result"
+        sql+=pageMissingDates();
+        if(overview)sql+=", summary_result AS ("+totals(null,false)+" FROM sku_result), top_owners AS (SELECT customer_code,physical,ROW_NUMBER() OVER (ORDER BY physical DESC,customer_code) ordinal FROM owner_totals WHERE physical IS NOT NULL ORDER BY physical DESC,customer_code LIMIT 10)";
+        // Statement-local only: share intermediate relations across report branches;
+        // do not change connection/global settings or any result/filter semantics.
+        sql+=" SELECT /*+ SET_VAR(optimizer_switch='derived_merge=off') SET_VAR(optimizer_search_depth=4) */ 'ROW' section,ordinal,"+jsonFields(fields)+" payload FROM page_missing"
             +" UNION ALL SELECT 'TOTAL' section,0 ordinal,JSON_OBJECT('total',COUNT(*)) payload FROM "+source+filter(p,owner);
         if(overview) {
             List<String> summary=new ArrayList<>(List.of("sku_count","unassessable","no_outbound_count","no_outbound_unknown"));
             for(String field:List.of("physical","available")){summary.add(field);summary.add(field+"_known");summary.add(field+"_unknown");}
             sql+=" UNION ALL SELECT 'OVERVIEW' section,0 ordinal,"+jsonFields(summary)+" payload FROM summary_result"
                 +" UNION ALL SELECT 'OWNER' section,ordinal,JSON_OBJECT('customer_code',customer_code,'physical',physical) payload FROM top_owners"
-                +" UNION ALL SELECT 'OWNER_META' section,0 ordinal,JSON_OBJECT('unknown_count',COUNT(*)) payload FROM owner_result WHERE physical IS NULL";
+                +" UNION ALL SELECT 'OWNER_META' section,0 ordinal,JSON_OBJECT('unknown_count',COUNT(*)) payload FROM owner_totals WHERE physical IS NULL";
         }
         return sql+" ORDER BY section,ordinal";
     }
diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/query/InventoryTurnoverContextTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/query/InventoryTurnoverContextTest.java
index 594492337..f4f64651e 100644
--- a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/query/InventoryTurnoverContextTest.java
+++ b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/query/InventoryTurnoverContextTest.java
@@ -35,7 +35,12 @@ class InventoryTurnoverContextTest {
         assertEquals(END.minusDays(29),first.analyticsContext().rangeStartDate());
         assertEquals(END.plusDays(1),first.analyticsContext().rangeEndDateExclusive());
         assertEquals(first,at(NOW.plusSeconds(60)).resolve(request(first.snapshotContext())).response());
-        verify(mapper,times(2)).samples("demo","W",IH,utc(NOW),END.minusDays(29),END.plusDays(1));
+        var page2=new InventoryTurnoverRequest("demo","W",END,30,"SKU",null,null,"ALL",null,null,"PHYSICAL_DESC",2,20,false,first.snapshotContext());
+        var second=at(NOW.plusSeconds(90)).resolve(page2);
+        assertEquals(first,second.response());
+        assertEquals(InventoryTurnoverSql.input(request(first.snapshotContext()),contexts.resolve(request(first.snapshotContext()))).days(),
+            InventoryTurnoverSql.input(page2,second).days());
+        verify(mapper,times(4)).samples("demo","W",IH,utc(NOW),END.minusDays(29),END.plusDays(1));
         // A subsequently published sample cannot appear in the old asOf query. If reconstruction
         // ever differs, reject the context instead of mixing old charts and new rows.
         when(mapper.samples(eq("demo"),eq("W"),eq(IH),any(),any(),any())).thenReturn(List.of());
diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/query/InventoryTurnoverSqlTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/query/InventoryTurnoverSqlTest.java
index 13018c274..af7bed0fe 100644
--- a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/query/InventoryTurnoverSqlTest.java
+++ b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/query/InventoryTurnoverSqlTest.java
@@ -103,6 +103,16 @@ class InventoryTurnoverSqlTest {
         assertNull(absent.get("w30_outbound"));assertNull(absent.get("w30_observed"));
         decimal("0",query("overview",input(q,30,0,true)).get(0).get("no_outbound_count"));
     }
+    @Test void warehouseChartUsesSalesUnitsAndCoverageIndependentlyOfStockProfiles() throws Exception {
+        seed("A","SKU",30);
+        execute("UPDATE ll_selfop_sku_profile SET quantity_unit='UNKNOWN'");
+        execute("UPDATE ll_selfop_inventory_snapshot SET physical_quantity=NULL");
+        var p=input(request("SKU","ALL","PHYSICAL_DESC",null,null,1,20),30,0,false);
+        var row=query("chartWindows",p).get(0);decimal("300",row.get("w30_outbound"));
+        execute("UPDATE ll_selfop_sku_daily_sales SET quantity_unit='UNKNOWN' WHERE business_date=DATE '"+END+"'");
+        row=query("chartWindows",p).get(0);assertNull(row.get("w30_outbound"));
+        decimal("290",row.get("w30_observed"));decimal("29",row.get("w30_complete"));decimal("6",row.get("w7_complete"));
+    }
     @Test void fullScopeFilterPrecedesPaginationAndKeepsSoldOutHistoricalCandidates() throws Exception {
         seed("A","LOW",30);seed("A","HIGH",30);seed("B","SOLD",30);
         execute("UPDATE ll_selfop_inventory_snapshot SET available_quantity=3000,physical_quantity=3000 WHERE sku_code='HIGH' AND snapshot_run_id=22 AND product_quality=0");
@@ -137,6 +147,231 @@ class InventoryTurnoverSqlTest {
         assertEquals(2,report.stream().filter(r->"OWNER".equals(r.get("section"))).count());
         assertEquals(1,report.stream().filter(r->"OVERVIEW".equals(r.get("section"))).count());
     }
+    @Test void frozenSamplesExcludeUnselectedVersionsWithoutDroppingMissingDays() throws Exception {
+        seed("A","SKU",30);
+        profile(999,"A","UNSELECTED");stock(999,"A","UNSELECTED",0,999,999);
+        var q=request("SKU","ALL","PHYSICAL_DESC",null,null,1,20);
+        var rows=reportRows(input(q,30,2,true));
+        assertEquals(1,rows.size());decimal("2",rows.get(0).get("sample_count"));
+        assertNull(rows.get(0).get("turnover"));
+        assertEquals(28,rows.get(0).get("missing_dates").toString().split(",").length);
+        rows=reportRows(input(q,30,0,true));
+        assertEquals(1,rows.size());decimal("0",rows.get(0).get("sample_count"));
+        assertEquals(30,rows.get(0).get("missing_dates").toString().split(",").length);
+        decimal("300",rows.get(0).get("w30_outbound"));
+    }
+    @Test void repeatedSampleRunAndHighestDateBitKeepAllThreeWindowsExact() throws Exception {
+        seed("A","SKU",30);
+        for(boolean onlyLast:List.of(true,false)) for(int n:List.of(7,15,30)) for(String group:List.of("SKU","OWNER")) {
+            var dates=new ArrayList<InventoryTurnoverSql.Day>();var missing=new ArrayList<String>();
+            for(int i=0;i<30;i++) {
+                var date=END.minusDays(29-i);boolean present=onlyLast?i==29:i!=29;
+                // Reusing a version on multiple frozen dates must retain each date.
+                dates.add(new InventoryTurnoverSql.Day(date,date.atStartOfDay().minusHours(8),1,present?100L:null));
+                if(i>=30-n&&!present)missing.add(date.toString());
+            }
+            var q=new InventoryTurnoverRequest("demo","W",END,n,group,null,null,"ALL",null,null,null,1,20,true,null);
+            var p=new InventoryTurnoverSql.Input(q,11,22,true,true,List.copyOf(dates));
+            var row=query("page",p).get(0);
+            decimal(onlyLast?"536870912":"536870911",row.get("sample_mask"));
+            decimal(String.valueOf(onlyLast?1:n-1),row.get("sample_count"));
+            assertEquals(String.join(",",missing),row.get("missing_dates"));
+            assertNull(row.get("average_physical"));assertNull(row.get("turnover"));
+            assertEquals(row.get("missing_dates"),reportRows(p).get(0).get("missing_dates"));
+        }
+    }
+    @Test void sparseSamplesKeepExactDateComplementAndFullOwnerDenominator() throws Exception {
+        seed("A","ONE",30);seed("A","TWO",30);
+        // Both edge bits and a middle bit, with different missing members per day.
+        execute("DELETE FROM ll_selfop_inventory_snapshot WHERE warehouse_code='W' AND ((sku_code='ONE' AND snapshot_run_id IN (100,114)) OR (sku_code='TWO' AND snapshot_run_id=129))");
+        for(int n:List.of(7,15,30)) {
+            var q=new InventoryTurnoverRequest("demo","W",END,n,"OWNER",null,null,"ALL",null,null,null,1,20,true,null);
+            var p=input(q,30,30,true);
+            var row=reportRows(p).get(0);
+            var expected=new ArrayList<LocalDate>();
+            for(int index:List.of(0,14,29)) if(index>=30-n) expected.add(END.minusDays(29-index));
+            decimal(String.valueOf(n-expected.size()),row.get("sample_count"));
+            assertEquals(String.join(",",expected.stream().map(LocalDate::toString).toList()),row.get("missing_dates"));
+            assertNull(row.get("average_physical"));assertNull(row.get("turnover"));
+            assertEquals(row.get("missing_dates"),query("page",p).get(0).get("missing_dates"));
+        }
+        var pageTwo=new InventoryTurnoverRequest("demo","W",END,30,"SKU",null,null,"ALL",null,null,null,2,1,false,null);
+        var second=reportRows(input(pageTwo,30,30,true)).get(0);
+        assertEquals("TWO",second.get("sku_code"));decimal("29",second.get("sample_count"));assertEquals(END.toString(),second.get("missing_dates"));
+        // A stock/profile-only candidate without samples invalidates all owner days.
+        profile(22,"A","NO-SAMPLES");stock(22,"A","NO-SAMPLES",0,0,0);
+        var owner=reportRows(input(request("OWNER","ALL","PHYSICAL_DESC",null,null,1,20),30,30,true)).get(0);
+        decimal("0",owner.get("sample_count"));assertEquals(30,owner.get("missing_dates").toString().split(",").length);
+    }
+    @Test void reportBindsExactLargeSampleVersionsAndKeepsWindowDenominators() throws Exception {
+        seed("A","SKU",30);
+        long first=9_007_199_254_740_993L;
+        for(int i=0;i<30;i++) {
+            execute("UPDATE ll_selfop_sku_profile SET profile_run_id="+(first+i)+" WHERE profile_run_id="+(100+i));
+            execute("UPDATE ll_selfop_inventory_snapshot SET snapshot_run_id="+(first+i)+",physical_quantity=CASE WHEN product_quality=0 THEN "+((i+1)*10)+" ELSE 0 END WHERE snapshot_run_id="+(100+i));
+        }
+        List<InventoryTurnoverSql.Day> days=new ArrayList<>();
+        for(int i=0;i<30;i++){var date=END.minusDays(29-i);days.add(new InventoryTurnoverSql.Day(date,date.atStartOfDay().minusHours(8),1,first+i));}
+        for(int n:List.of(7,15,30)) {
+            var q=new InventoryTurnoverRequest("demo","W",END,n,"SKU",null,null,"ALL",null,null,null,1,20,true,null);
+            var p=new InventoryTurnoverSql.Input(q,11,22,true,true,List.copyOf(days));
+            var row=reportRows(p).get(0);
+            decimal(String.valueOf(n),row.get("sample_count"));
+            decimal(n==7?"270":n==15?"230":"155",row.get("average_physical"));
+            decimal("70",row.get("w7_outbound"));decimal("150",row.get("w15_outbound"));decimal("300",row.get("w30_outbound"));
+            assertNotNull(row.get("turnover"));
+        }
+        var missing=days.get(14);days.set(14,new InventoryTurnoverSql.Day(missing.date(),missing.startUtc(),1,null));
+        var row=reportRows(new InventoryTurnoverSql.Input(request("SKU","ALL","PHYSICAL_DESC",null,null,1,20),11,22,true,true,List.copyOf(days))).get(0);
+        decimal("29",row.get("sample_count"));assertNull(row.get("turnover"));
+        assertEquals(missing.date().toString(),row.get("missing_dates"));
+    }
+    @Test void reportPaginationKeepsTotalAndCanOmitOverview() throws Exception {
+        seed("A","SKU",30);seed("B","SKU",30);
+        var first=reportRows(input(request("SKU","ALL","PHYSICAL_DESC",null,null,1,1),30,30,true));
+        var secondQuery=new InventoryTurnoverRequest("demo","W",END,30,"SKU",null,null,"ALL",null,null,"PHYSICAL_DESC",2,1,false,"frozen");
+        var second=query("report",input(secondQuery,30,30,true));
+        var row=second.stream().filter(r->"ROW".equals(r.get("section"))).findFirst().orElseThrow();
+        assertNotEquals(first.get(0).get("customer_code"),payload(row).path("customer_code").asText());
+        assertEquals(2,second.size());
+        var beyond=new InventoryTurnoverRequest("demo","W",END,30,"SKU",null,null,"ALL",null,null,"PHYSICAL_DESC",3,1,false,"frozen");
+        var report=query("report",input(beyond,30,30,true));assertEquals(1,report.size());
+        assertEquals("TOTAL",report.get(0).get("section"));assertEquals(2,payload(report.get(0)).path("total").asInt());
+    }
+    @Test void projectedFormalWindowsKeepPartialObservationsAndUnknownDistinctFromZero() throws Exception {
+        seed("A","SKU",30);
+        for(String group:List.of("SKU","OWNER")) {
+            var p=input(request(group,"ALL","PHYSICAL_DESC",null,null,1,20),0,30,true);
+            var row=reportRows(p).get(0);
+            assertNull(row.get("w30_outbound"));decimal("300",row.get("w30_observed"));decimal("30",row.get("w30_partial"));
+        }
+        execute("DELETE FROM ll_selfop_sku_daily_sales WHERE warehouse_code='W'");
+        for(String group:List.of("SKU","OWNER")) {
+            var p=input(request(group,"ALL","PHYSICAL_DESC",null,null,1,20),0,30,true);
+            var unknown=reportRows(p).get(0);assertNull(unknown.get("w30_outbound"));assertNull(unknown.get("w30_observed"));
+            var partialDays=p.days().stream().map(d->new InventoryTurnoverSql.Day(d.date(),d.startUtc(),2,d.sampleRun())).toList();
+            var partial=reportRows(new InventoryTurnoverSql.Input(p.query(),11,22,true,true,partialDays)).get(0);
+            assertNull(partial.get("w30_outbound"));decimal("0",partial.get("w30_observed"));decimal("30",partial.get("w30_partial"));
+            var complete=reportRows(input(p.query(),30,30,true)).get(0);
+            decimal("0",complete.get("w30_outbound"));decimal("30",complete.get("w30_complete"));
+        }
+    }
+    @Test void zeroEvidenceUsesExactBeijingMidnightIncludingMilliseconds() throws Exception {
+        seed("A","SKU",0);execute("DELETE FROM ll_selfop_sku_daily_sales WHERE warehouse_code='W'");
+        for(int n:List.of(7,15,30)) for(int millis:List.of(-1,0,1)) {
+            LocalDate start=END.minusDays(n-1);var utc=start.atStartOfDay().minusHours(8).plusNanos(millis*1_000_000L);
+            execute("UPDATE ll_selfop_sku_profile SET earliest_evidence_at_utc=TIMESTAMP '"+utc.toString().replace('T',' ')+"'");
+            var p=input(request("SKU","ALL","PHYSICAL_DESC",null,null,1,20),30,0,true);
+            var row=reportRows(p).get(0);decimal(String.valueOf(n-(millis>0?1:0)),row.get("w"+n+"_complete"));
+            if(millis>0)assertNull(row.get("w"+n+"_outbound"));else decimal("0",row.get("w"+n+"_outbound"));
+            execute("INSERT INTO ll_selfop_sku_daily_sales VALUES(11,'demo','W','A','SKU',DATE '"+start+"',0,'PCS')");
+            row=reportRows(p).get(0);decimal(String.valueOf(n),row.get("w"+n+"_complete"));decimal("0",row.get("w"+n+"_outbound"));
+            execute("DELETE FROM ll_selfop_sku_daily_sales WHERE warehouse_code='W'");
+        }
+    }
+    @Test void sparseCorrectionRevokesInvalidActualZeroAndHandlesEvidenceAfterLastBoundary() throws Exception {
+        seed("A","SKU",0);execute("DELETE FROM ll_selfop_sku_daily_sales WHERE warehouse_code='W'");
+        var p=input(request("SKU","ALL","PHYSICAL_DESC",null,null,1,20),1,0,true);
+        var afterLast=END.atStartOfDay().minusHours(8).plusNanos(1_000_000);
+        execute("UPDATE ll_selfop_sku_profile SET earliest_evidence_at_utc=TIMESTAMP '"+afterLast.toString().replace('T',' ')+"'");
+        var row=reportRows(p).get(0);decimal("0",row.get("w7_complete"));assertNull(row.get("w7_observed"));decimal("0",row.get("w7_partial"));
+        execute("INSERT INTO ll_selfop_sku_daily_sales VALUES(11,'demo','W','A','SKU',DATE '"+END+"',1.234567,'PCS')");
+        row=reportRows(p).get(0);decimal("1",row.get("w7_complete"));decimal("1.234567",row.get("w7_observed"));
+        execute("UPDATE ll_selfop_sku_profile SET earliest_evidence_at_utc=TIMESTAMP '2026-01-01 00:00:00'");
+        execute("UPDATE ll_selfop_sku_daily_sales SET quantity_unit='UNKNOWN' WHERE warehouse_code='W'");
+        row=reportRows(p).get(0);decimal("0",row.get("w7_complete"));assertNull(row.get("w7_observed"));decimal("0",row.get("w7_partial"));
+    }
+    @Test void presentInvalidSamplesNeverQualifyAndOwnerZeroNeedsEverySku() throws Exception {
+        seed("A","ONE",30);seed("A","TWO",30);
+        String key=" WHERE snapshot_run_id=100 AND sku_code='TWO' AND product_quality=0";
+        for(String mutation:List.of("physical_quantity=NULL","physical_quantity=-1","product_quality=9","quality_status='UNKNOWN',quality_reason='UNCONFIRMED'","profile")) {
+            if("profile".equals(mutation))execute("UPDATE ll_selfop_sku_profile SET quantity_unit='UNKNOWN' WHERE profile_run_id=100 AND sku_code='TWO'");
+            else execute("UPDATE ll_selfop_inventory_snapshot SET "+mutation+key);
+            var owner=reportRows(input(request("OWNER","ALL","PHYSICAL_DESC",null,null,1,20),30,30,true)).get(0);
+            decimal("29",owner.get("sample_count"));assertNull(owner.get("turnover"));assertEquals(END.minusDays(29).toString(),owner.get("missing_dates"));
+            execute("DELETE FROM ll_selfop_inventory_snapshot WHERE snapshot_run_id=100 AND sku_code='TWO'");stock(100,"A","TWO",0,450,450);stock(100,"A","TWO",1,50,0);
+            execute("UPDATE ll_selfop_sku_profile SET quantity_unit='PCS' WHERE profile_run_id=100 AND sku_code='TWO'");
+        }
+        execute("UPDATE ll_selfop_inventory_snapshot SET physical_quantity=0 WHERE snapshot_run_id>=100");
+        var owner=reportRows(input(request("OWNER","ALL","PHYSICAL_DESC",null,null,1,20),30,30,true)).get(0);
+        decimal("30",owner.get("sample_count"));decimal("0",owner.get("average_physical"));assertNull(owner.get("turnover"));
+    }
+    @Test void ownerInvalidActualAndAbsentMemberKeepCoveredZeroDifferentFromUnknown() throws Exception {
+        seed("A","ONE",0);seed("A","TWO",0);execute("DELETE FROM ll_selfop_sku_daily_sales WHERE warehouse_code='W'");
+        execute("INSERT INTO ll_selfop_sku_daily_sales VALUES(11,'demo','W','A','ONE',DATE '"+END+"',10,'UNKNOWN')");
+        var q=request("OWNER","ALL","PHYSICAL_DESC",null,null,1,20);
+        var row=reportRows(input(q,30,0,true)).get(0);decimal("6",row.get("w7_complete"));decimal("0",row.get("w7_observed"));decimal("1",row.get("w7_partial"));
+        row=reportRows(input(q,0,0,true)).get(0);assertNull(row.get("w7_observed"));decimal("0",row.get("w7_partial"));
+        execute("INSERT INTO ll_selfop_sku_daily_sales VALUES(11,'demo','W','A','TWO',DATE '"+END+"',10,'UNKNOWN')");
+        row=reportRows(input(q,1,0,true)).get(0);assertNull(row.get("w7_observed"));decimal("0",row.get("w7_complete"));decimal("0",row.get("w7_partial"));
+    }
+    @Test void decimalWindowsRemainExactBeyondJavascriptIntegerRangeWithoutDuplicatingCandidates() throws Exception {
+        for(String sku:List.of("ONE","TWO","THREE","FOUR"))seed("A",sku,30);
+        String amount="99999999999999.123456";var quantity=new BigDecimal(amount);
+        execute("UPDATE ll_selfop_sku_daily_sales SET outbound_quantity="+amount+" WHERE warehouse_code='W'");
+        execute("UPDATE ll_selfop_inventory_snapshot SET physical_quantity=CASE WHEN product_quality=0 THEN "+amount+" ELSE 0 END");
+        var report=query("report",input(request("OWNER","ALL","PHYSICAL_DESC",null,null,1,20),30,30,true));
+        var row=payload(report.stream().filter(r->"ROW".equals(r.get("section"))).findFirst().orElseThrow());
+        decimal(quantity.multiply(BigDecimal.valueOf(120)).toPlainString(),row.path("w30_outbound").decimalValue());
+        decimal(quantity.multiply(BigDecimal.valueOf(4)).toPlainString(),row.path("average_physical").decimalValue());
+        decimal("30",row.path("turnover").decimalValue());assertEquals(30,row.path("sample_count").asInt());
+        var skuReport=query("report",input(request("SKU","ALL","PHYSICAL_DESC",null,null,1,20),30,30,true));
+        assertEquals(4,skuReport.stream().filter(r->"ROW".equals(r.get("section"))).count());
+        for(var raw:skuReport)if("ROW".equals(raw.get("section")))decimal(quantity.multiply(BigDecimal.valueOf(30)).toPlainString(),payload(raw).path("w30_outbound").decimalValue());
+    }
+    @Test void threeSourceCandidateUnionKeepsFullOwnerDenominator() throws Exception {
+        profile(100,"A","HISTORY");profile(11,"A","HISTORY");profile(22,"A","HISTORY");
+        stock(22,"A","STOCK",0,0,0);
+        execute("INSERT INTO ll_selfop_sku_daily_sales VALUES(11,'demo','W','A','SALES',DATE '"+END+"',1.123456,'PCS')");
+        var skuRows=reportRows(input(request("SKU","ALL","PHYSICAL_DESC",null,null,1,20),30,30,true));
+        assertEquals(Set.of("HISTORY","STOCK","SALES"),new HashSet<>(skuRows.stream().map(r->r.get("sku_code")).toList()));
+        assertEquals(3,skuRows.size());
+        var report=query("report",input(request("OWNER","ALL","PHYSICAL_DESC",null,null,1,20),30,30,true));
+        var row=payload(report.stream().filter(r->"ROW".equals(r.get("section"))).findFirst().orElseThrow());
+        decimal("1.123456",row.path("w30_outbound").decimalValue());
+        assertEquals(0,row.path("sample_count").asInt());assertTrue(row.path("average_physical").isNull());
+        assertEquals(30,row.path("missing_dates").asText().split(",").length);
+    }
+    @Test void ownerStockChartSurvivesMissingWindowsAndEmptySkuFilter() throws Exception {
+        seed("A","SKU",0);seed("B","SKU",0);seed("C","SOLD",0);
+        execute("UPDATE ll_selfop_inventory_snapshot SET physical_quantity=NULL,quality_status='UNKNOWN',quality_reason='FIELD_MISSING' WHERE customer_code='B' AND snapshot_run_id=22");
+        execute("UPDATE ll_selfop_inventory_snapshot SET physical_quantity=0,available_quantity=0 WHERE customer_code='C' AND snapshot_run_id=22");
+        var baseline=query("report",input(request("SKU","ALL","PHYSICAL_DESC",null,null,1,20),7,0,true));
+        var baselineSummary=payload(baseline.stream().filter(r->"OVERVIEW".equals(r.get("section"))).findFirst().orElseThrow());
+        for(String group:List.of("SKU","OWNER")) {
+            var full=input(request(group,"ALL","PHYSICAL_DESC",null,null,1,20),7,0,true);
+            var standalone=query("page",full);var reportRows=reportRows(full);
+            assertEquals(standalone.size(),reportRows.size());
+            for(int i=0;i<standalone.size();i++) for(var field:reportRows.get(i).entrySet()) {
+                Object expected=standalone.get(i).get(field.getKey()),actual=field.getValue();
+                if(expected instanceof Number)decimal(expected.toString(),actual);
+                else if(expected instanceof java.sql.Date)assertEquals(expected.toString(),actual);
+                else assertEquals(expected,actual,field.getKey());
+            }
+            var q=request(group,"ALL","PHYSICAL_DESC","SKU".equals(group)?"not-present":null,null,1,20);
+            var report=query("report",input(q,7,0,true));
+            var chart=report.stream().filter(r->"OWNER".equals(r.get("section"))).toList();
+            assertEquals(2,chart.size());
+            var known=payload(chart.get(0));assertEquals("A",known.path("customer_code").asText());decimal("610",known.path("physical").decimalValue());
+            var unknown=report.stream().filter(r->"OWNER_META".equals(r.get("section"))).findFirst().orElseThrow();
+            assertEquals(1,payload(unknown).path("unknown_count").asInt());
+            var summary=report.stream().filter(r->"OVERVIEW".equals(r.get("section"))).findFirst().orElseThrow();
+            assertEquals(3,payload(summary).path("sku_count").asInt());assertEquals(baselineSummary,payload(summary));
+            var total=report.stream().filter(r->"TOTAL".equals(r.get("section"))).findFirst().orElseThrow();
+            assertEquals("SKU".equals(group)?0:3,payload(total).path("total").asInt());
+            assertEquals("SKU".equals(group)?0:3,report.stream().filter(r->"ROW".equals(r.get("section"))).count());
+        }
+    }
+    private static com.fasterxml.jackson.databind.JsonNode payload(Map<String,Object> row) throws Exception {
+        var json=new com.fasterxml.jackson.databind.ObjectMapper().enable(com.fasterxml.jackson.databind.DeserializationFeature.USE_BIG_DECIMAL_FOR_FLOATS);Object raw=row.get("payload");
+        return raw instanceof byte[] bytes?json.readTree(bytes):json.readTree(raw.toString());
+    }
+    private List<Map<String,Object>> reportRows(InventoryTurnoverSql.Input p) throws Exception {
+        var result=new ArrayList<Map<String,Object>>();var json=new com.fasterxml.jackson.databind.ObjectMapper();
+        for(var row:query("report",p)) if("ROW".equals(row.get("section")))
+            result.add(json.convertValue(payload(row),new com.fasterxml.jackson.core.type.TypeReference<Map<String,Object>>(){}));
+        return result;
+    }
     private void seed(String owner,String sku,int samples) throws Exception {
         profile(11,owner,sku);profile(22,owner,sku);stock(22,owner,sku,0,600,600);stock(22,owner,sku,1,10,0);
         for(int i=0;i<30;i++) {
@@ -159,6 +394,10 @@ class InventoryTurnoverSqlTest {
             Object value=meta.getValue(parameter.getProperty());String literal=value==null?"NULL":value instanceof Boolean b?b?"1":"0":value instanceof Number?value.toString():"'"+value.toString().replace("'","''")+"'";
             sql=sql.replaceFirst("\\?",java.util.regex.Matcher.quoteReplacement(literal));
         }
+        // MySQL SIGNED is a signed 64-bit integer; H2's alias is only 32-bit.
+        sql=sql.replace(" AS SIGNED)"," AS BIGINT)");
+        // MySQL's unbounded CHAR cast preserves the date; H2 defaults to CHAR(1).
+        sql=sql.replace("CAST(business_date AS CHAR)","CAST(business_date AS VARCHAR(10))");
         // H2 inlines repeatedly referenced CTEs exponentially. Materialize each unchanged
         // production relation in this local fixture; MySQL execution plans remain unverified.
         List<String> relations=new ArrayList<>();int cursor=5;