fix: stock analytics report shows incorrect data there's no stock movement in a period

erpnext/stock/report/stock_analytics/stock_analytics.py

@@ -1,6 +1,7 @@
 # Copyright (c) 2013, Frappe Technologies Pvt. Ltd. and contributors
 # For license information, please see license.txt
 import datetime
+from typing import List
 
 import frappe
 from frappe import _, scrub
@@ -148,18 +149,26 @@ def get_periodic_data(entry, filters):
 	                        - Warehouse A : bal_qty/value
 	                        - Warehouse B : bal_qty/value
 	"""
+
+	expected_ranges = get_period_date_ranges(filters)
+	expected_periods = []
+	for _start_date, end_date in expected_ranges:
+		expected_periods.append(get_period(end_date, filters))
+
 	periodic_data = {}
 	for d in entry:
 		period = get_period(d.posting_date, filters)
 		bal_qty = 0
 
+		fill_intermediate_periods(periodic_data, d.item_code, period, expected_periods)
+
 		# if period against item does not exist yet, instantiate it
 		# insert existing balance dict against period, and add/subtract to it
 		if periodic_data.get(d.item_code) and not periodic_data.get(d.item_code).get(period):
 			previous_balance = periodic_data[d.item_code]["balance"].copy()
 			periodic_data[d.item_code][period] = previous_balance
 
-		if d.voucher_type == "Stock Reconciliation":
+		if d.voucher_type == "Stock Reconciliation" and not d.batch_no:
 			if periodic_data.get(d.item_code) and periodic_data.get(d.item_code).get("balance").get(
 				d.warehouse
 			):
@@ -186,6 +195,36 @@ def get_periodic_data(entry, filters):
 	return periodic_data
 
 
+def fill_intermediate_periods(
+	periodic_data, item_code: str, current_period: str, all_periods: List[str]
+) -> None:
+	"""There might be intermediate periods where no stock ledger entry exists, copy previous previous data.
+
+	Previous data is ONLY copied if period falls in report range and before period being processed currently.
+
+	args:
+	        current_period: process till this period (exclusive)
+	        all_periods: all periods expected in report via filters
+	        periodic_data: report's periodic data
+	        item_code: item_code being processed
+	"""
+
+	previous_period_data = None
+	for period in all_periods:
+		if period == current_period:
+			return
+
+		if (
+			periodic_data.get(item_code)
+			and not periodic_data.get(item_code).get(period)
+			and previous_period_data
+		):
+			# This period should exist since it's in report range, assign previous period data
+			periodic_data[item_code][period] = previous_period_data.copy()
+
+		previous_period_data = periodic_data.get(item_code, {}).get(period)
+
+
 def get_data(filters):
 	data = []
 	items = get_items(filters)
@@ -194,6 +233,8 @@ def get_data(filters):
 	periodic_data = get_periodic_data(sle, filters)
 	ranges = get_period_date_ranges(filters)
 
+	today = getdate()
+
 	for dummy, item_data in item_details.items():
 		row = {
 			"name": item_data.name,
@@ -202,14 +243,15 @@ def get_data(filters):
 			"uom": item_data.stock_uom,
 			"brand": item_data.brand,
 		}
-		total = 0
-		for dummy, end_date in ranges:
+		previous_period_value = 0.0
+		for start_date, end_date in ranges:
 			period = get_period(end_date, filters)
 			period_data = periodic_data.get(item_data.name, {}).get(period)
-			amount = sum(period_data.values()) if period_data else 0
-			row[scrub(period)] = amount
-			total += amount
-		row["total"] = total
+			if period_data:
+				row[scrub(period)] = previous_period_value = sum(period_data.values())
+			else:
+				row[scrub(period)] = previous_period_value if today >= start_date else None
+
 		data.append(row)
 
 	return data

erpnext/stock/report/stock_analytics/test_stock_analytics.py

@@ -1,13 +1,59 @@
 import datetime
 
+import frappe
 from frappe import _dict
 from frappe.tests.utils import FrappeTestCase
+from frappe.utils.data import add_to_date, get_datetime, getdate, nowdate
 
 from erpnext.accounts.utils import get_fiscal_year
-from erpnext.stock.report.stock_analytics.stock_analytics import get_period_date_ranges
+from erpnext.stock.doctype.item.test_item import make_item
+from erpnext.stock.doctype.stock_entry.stock_entry_utils import make_stock_entry
+from erpnext.stock.report.stock_analytics.stock_analytics import execute, get_period_date_ranges
+
+
+def stock_analytics(filters):
+	col, data, *_ = execute(filters)
+	return col, data
 
 
 class TestStockAnalyticsReport(FrappeTestCase):
+	def setUp(self) -> None:
+		self.item = make_item().name
+		self.warehouse = "_Test Warehouse - _TC"
+
+	def assert_single_item_report(self, movement, expected_buckets):
+		self.generate_stock(movement)
+		filters = _dict(
+			range="Monthly",
+			from_date=movement[0][1].replace(day=1),
+			to_date=movement[-1][1].replace(day=28),
+			value_quantity="Quantity",
+			company="_Test Company",
+			item_code=self.item,
+		)
+
+		cols, data = stock_analytics(filters)
+
+		self.assertEqual(len(data), 1)
+		row = frappe._dict(data[0])
+		self.assertEqual(row.name, self.item)
+		self.compare_analytics_row(row, cols, expected_buckets)
+
+	def generate_stock(self, movement):
+		for qty, posting_date in movement:
+			args = {"item": self.item, "qty": abs(qty), "posting_date": posting_date}
+			args["to_warehouse" if qty > 0 else "from_warehouse"] = self.warehouse
+			make_stock_entry(**args)
+
+	def compare_analytics_row(self, report_row, columns, expected_buckets):
+		# last (N) cols will be monthly data
+		no_of_buckets = len(expected_buckets)
+		month_cols = [col["fieldname"] for col in columns[-no_of_buckets:]]
+
+		actual_buckets = [report_row.get(col) for col in month_cols]
+
+		self.assertEqual(actual_buckets, expected_buckets)
+
 	def test_get_period_date_ranges(self):
 
 		filters = _dict(range="Monthly", from_date="2020-12-28", to_date="2021-02-06")
@@ -33,3 +79,38 @@ def test_get_period_date_ranges_yearly(self):
 		]
 
 		self.assertEqual(ranges, expected_ranges)
+
+	def test_basic_report_functionality(self):
+		"""Stock analytics report generates balance "as of" periods based on
+		user defined ranges. Check that this behaviour is correct."""
+
+		# create stock movement in 3 months at 15th of month
+		today = getdate()
+		movement = [
+			(10, add_to_date(today, months=0).replace(day=15)),
+			(-5, add_to_date(today, months=1).replace(day=15)),
+			(10, add_to_date(today, months=2).replace(day=15)),
+		]
+		self.assert_single_item_report(movement, [10, 5, 15])
+
+	def test_empty_month_in_between(self):
+		today = getdate()
+		movement = [
+			(100, add_to_date(today, months=0).replace(day=15)),
+			(-50, add_to_date(today, months=1).replace(day=15)),
+			# Skip a month
+			(20, add_to_date(today, months=3).replace(day=15)),
+		]
+		self.assert_single_item_report(movement, [100, 50, 50, 70])
+
+	def test_multi_month_missings(self):
+		today = getdate()
+		movement = [
+			(100, add_to_date(today, months=0).replace(day=15)),
+			(-50, add_to_date(today, months=1).replace(day=15)),
+			# Skip a month
+			(20, add_to_date(today, months=3).replace(day=15)),
+			# Skip another month
+			(-10, add_to_date(today, months=5).replace(day=15)),
+		]
+		self.assert_single_item_report(movement, [100, 50, 50, 70, 70, 60])