""" Generate dummy CSV files for data import testing. CSV fields are based on README.md requirements. Generates realistic data with proper inventory flow and behavior distribution. """ import os import csv import random import math import datetime from collections import defaultdict try: from faker import Faker except ImportError: print("Faker is not installed. Please run: pip install faker") exit(1) # Initialize Faker fake = Faker() # Base Configuration SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) SMART_INVENTORY_DIR = os.path.dirname(os.path.dirname(SCRIPT_DIR)) OUTPUT_DIR = os.path.join(SMART_INVENTORY_DIR, "dummy_data", "data", "csv_upload") os.makedirs(OUTPUT_DIR, exist_ok=True) # ============================================================ # BEHAVIOR CONFIGURATION (from generate_dummy_data.py) # ============================================================ BEHAVIOR_ASSIGNMENT_ORDER = ["normal", "slow", "dead"] BEHAVIOR_CONFIG = { "default_ratio": { "normal": 0.81, # 81% normal (includes in_stock, low_stock, out_of_stock, overstock) "slow": 0.14, # 14% slow movers "dead": 0.05, # 5% dead stock }, "small_dataset_threshold": 10, } # Stock status distribution within NORMAL products (not slow/dead) # Target: 60% in_stock, 8% low_stock, 3% out_of_stock, 10% overstock STOCK_STATUS_RATIOS = { "in_stock": 0.74, # Good stock levels "low_stock": 0.10, # Below reorder point "out_of_stock": 0.04, # Zero stock "overstock": 0.12, # Excess stock } # ADS (Average Daily Sales) ranges by behavior type ADS_CONFIG = { "normal": {"min": 2, "max": 15}, # Normal products sell 2-15 units/day "slow": {"min": 0, "max": 0.5}, # Slow movers: 0-0.5 units/day (rarely sell) "dead": {"min": 0, "max": 0}, # Dead stock: no sales } # DOH (Days on Hand) targets DOH_CONFIG = { "normal": {"target": 30, "max": 60}, # Normal: 30 days target, max 60 "slow": {"target": 90, "max": 180}, # Slow: higher DOH expected "dead": {"target": 180, "max": 365}, # Dead: very high DOH } # Sample data pools CATEGORIES = [ "Better Homes & Gardens", "Luggage", "Paint Brush", "Automobile", "COTTON", "Women's Wear", "Coffee", "Soda", "Wallet Category", "Natural Cheese", "Women's", "Perfumes", "Skincare", "Chocolates", "Decorative", "Milk Prod", "Chicken", "Men's Wear", "Electronics", "Home Decor", "Kitchen Appliances" ] PRODUCT_NAMES = [ "Better Homes & Gardens Stemless Cocktail Glass", "3 Trolley Set", "Acrylic Paint Brush", "AI Translation Earbuds Real Time", "Birdfy Smart Bird Feeder with Camera", "Black Recycled Plastic Hopper Wild Bird Feeder", "Bundle Swagger", "BURBERRY Her Eau de Parfum Intense", "Cadbury Chocolate", "Chicken Sausage", "Better Homes & Gardens River Crest Bookcase", "Box of Treats Mini Perfume Sampler", "Baggy Shirt", "Boho bag-multicolor", "Black Gel Pen", "Candle Set", "Candy Pack", "Premium Coffee Blend", "Organic Green Tea", "Leather Wallet", "Sports Watch", "Wireless Earbuds", "Kitchen Mixer", "Floor Lamp", "Decorative Vase", "Smart Thermostat", "Bluetooth Speaker", "Running Shoes", "Yoga Mat", "Protein Powder" ] BRANDS = ["Gucci", "Bellavita", "BURBERRY", "LACOSTE", "Cadbury", "VIP", "Kellogg Company", "Lipton", "Boat-X", "New Brand", "Nike", "Adidas", "Samsung", "Apple", "Sony", None] LOCATION_NAMES = [ "Main Store", "Downtown Branch", "Mall Outlet", "Airport Store", "Warehouse Central", "North Point", "South Center", "East Wing", "West Side Store", "Online Fulfillment Center", "Suburban Plaza", "City Center", "Industrial Park", "Express Outlet", "Premium Store" ] VENDOR_NAMES = [ "Global Supplies Inc", "Premium Distributors", "FastTrack Logistics", "Quality Goods Ltd", "Metro Wholesale", "City Vendors Co", "Express Trading", "National Distributors", "Regional Suppliers", "Direct Source Ltd", "Prime Wholesale", "Elite Supplies", "Quick Ship Co" ] def build_behavior_pool(num_items: int) -> list: """Create a shuffled list of normal/slow/dead assignments based on desired ratios.""" if num_items <= 0: return [] ratio_map = BEHAVIOR_CONFIG["default_ratio"].copy() raw_counts = {} fractional = [] assigned_total = 0 for behavior in BEHAVIOR_ASSIGNMENT_ORDER: value = ratio_map.get(behavior, 0.0) * num_items count = int(math.floor(value)) raw_counts[behavior] = count assigned_total += count fractional.append((value - count, behavior)) remaining = num_items - assigned_total fractional.sort(key=lambda x: x[0], reverse=True) idx = 0 while remaining > 0: behavior = fractional[idx % len(fractional)][1] raw_counts[behavior] += 1 remaining -= 1 idx += 1 pool = [] for behavior in BEHAVIOR_ASSIGNMENT_ORDER: pool.extend([behavior] * raw_counts.get(behavior, 0)) random.shuffle(pool) return pool def get_custom_config(): """Get configuration from user input - simplified to essentials only""" print("\n" + "=" * 50) print("CSV Dummy Data Generator (Realistic Mode)") print("=" * 50) print("\nThis will generate realistic inventory data with:") print(" - 81% normal products, 14% slow movers, 5% dead stock") print(" - Proper inventory flow (purchases → stock → sales)") print(" - Evenly distributed transactions across time period") print(" - Positive on-hand quantities with realistic DOH") print("=" * 50) config = {} while True: try: days = int(input("\nEnter number of days of history: ").strip()) if days > 0: config['days'] = days break print("Please enter a positive number.") except ValueError: print("Invalid input. Please enter a number.") while True: try: products = int(input("Enter number of products: ").strip()) if products > 0: config['products'] = products break print("Please enter a positive number.") except ValueError: print("Invalid input. Please enter a number.") while True: try: locations = int(input("Enter number of locations: ").strip()) if locations > 0: config['locations'] = min(locations, len(LOCATION_NAMES)) break print("Please enter a positive number.") except ValueError: print("Invalid input. Please enter a number.") while True: try: vendors = int(input("Enter number of vendors: ").strip()) if vendors > 0: config['vendors'] = min(vendors, len(VENDOR_NAMES)) break print("Please enter a positive number.") except ValueError: print("Invalid input. Please enter a number.") return config def generate_sku(brand, product_idx): """Generate a realistic SKU""" if random.random() < 0.8: brand_prefix = (brand[:3].upper() if brand else "GEN") return f"{brand_prefix}-{random.randint(1000, 9999)}-{product_idx:03d}" return "" def generate_products(config, locations, vendors): """Generate product CSV data with location and vendor assignments""" products = [] num_products = config['products'] for i in range(1, num_products + 1): base_name = random.choice(PRODUCT_NAMES) variation = random.choice(["", "Pack", "Set", "Large", "Small", "Medium", "Classic", "Premium"]) product_name = f"{base_name} {variation}".strip() if variation else base_name # Make product names unique by adding index if needed product_name = f"{product_name} #{i}" if random.random() > 0.5 else product_name brand = random.choice(BRANDS) category = random.choice(CATEGORIES) short_name = product_name[:20] if len(product_name) > 20 else "" description = fake.sentence(nb_words=10) if random.random() > 0.3 else "" sku = generate_sku(brand, i) # Assign 1-3 locations per product num_locs = random.randint(1, min(3, len(locations))) product_locations = random.sample(locations, num_locs) location_name = "|".join(product_locations) # Assign 1-2 vendors per product num_vends = random.randint(1, min(2, len(vendors))) product_vendors = random.sample(vendors, num_vends) vendor_name = "|".join(product_vendors) products.append({ "product_name": product_name, "short_name": short_name, "description": description, "brand_name": brand if brand else "", "sku": sku, "category_name": category, "location_name": location_name, "vendor_name": vendor_name }) return products def simulate_inventory_flow(config, products, locations, vendors, start_date): """ Simulate realistic inventory flow day-by-day. Returns: purchase_orders, purchase_receives, sales_orders, sales_returns, purchase_returns, stock_transfers """ today = datetime.datetime.now() # Build product-location combinations with behavior assignment product_locations = [] for product in products: prod_locs = product['location_name'].split("|") prod_vendors = product['vendor_name'].split("|") for loc in prod_locs: product_locations.append({ "product": product, "location": loc, "vendors": prod_vendors }) # Assign behavior (normal/slow/dead) to each product-location num_pl = len(product_locations) behavior_pool = build_behavior_pool(num_pl) for i, pl in enumerate(product_locations): pl["behavior"] = behavior_pool[i] if i < len(behavior_pool) else "normal" # Assign ADS based on behavior ads_range = ADS_CONFIG[pl["behavior"]] pl["target_ads"] = random.uniform(ads_range["min"], ads_range["max"]) # Assign stock status for normal products if pl["behavior"] == "normal": rand = random.random() cumulative = 0 for status, ratio in STOCK_STATUS_RATIOS.items(): cumulative += ratio if rand <= cumulative: pl["stock_status"] = status break else: pl["stock_status"] = "in_stock" else: pl["stock_status"] = "slow" if pl["behavior"] == "slow" else "dead" # Print behavior distribution behavior_counts = defaultdict(int) for pl in product_locations: behavior_counts[pl["behavior"]] += 1 print(f"\nProduct-Location Behavior Distribution:") for behavior in BEHAVIOR_ASSIGNMENT_ORDER: count = behavior_counts[behavior] pct = (count / num_pl * 100) if num_pl > 0 else 0 print(f" - {behavior.capitalize()}: {count} ({pct:.1f}%)") # Initialize inventory tracking inventory = defaultdict(lambda: {"on_hand": 0, "total_received": 0, "total_sold": 0}) # Data containers purchase_orders = [] purchase_receives = [] sales_orders = [] sales_returns = [] purchase_returns = [] stock_transfers = [] # Counters po_counter = 1 pr_counter = 1 so_counter = 1 sr_counter = 1 vr_counter = 1 st_counter = 1 # Calculate initial stock needs based on expected sales print("\nPhase 1: Generating initial purchase orders and receives...") for pl in product_locations: product = pl["product"] location = pl["location"] vendor = random.choice(pl["vendors"]) key = (product["product_name"], location) # Calculate how much stock we need based on behavior expected_daily_sales = pl["target_ads"] # Initial stock = enough for DOH target + buffer doh_target = DOH_CONFIG[pl["behavior"]]["target"] if pl["behavior"] == "dead": # Dead stock: receive once, never sell initial_qty = random.randint(50, 200) elif pl["behavior"] == "slow": # Slow movers: limited initial stock initial_qty = random.randint(20, 80) else: # Normal products: stock based on expected sales initial_qty = int(expected_daily_sales * doh_target * random.uniform(1.0, 1.5)) initial_qty = max(initial_qty, 20) # Minimum stock # Create initial purchase order (at start of period) initial_date = start_date + datetime.timedelta(days=random.randint(0, 5)) po_ref = f"PO-{initial_date.strftime('%Y%m%d')}-{po_counter:04d}" unit_cost = round(random.uniform(5.0, 300.0), 2) purchase_orders.append({ "product_name": product["product_name"], "purchase_order_ref_no": po_ref, "vendor_name": vendor, "location_name": location, "purchase_order_date": initial_date.strftime("%Y-%m-%d"), "order_qty": initial_qty, "unit_cost": unit_cost }) # Create purchase receive (1-5 days after order) receive_date = initial_date + datetime.timedelta(days=random.randint(1, 5)) if receive_date > today: receive_date = today pr_ref = f"PR-{receive_date.strftime('%Y%m%d')}-{pr_counter:04d}" receive_qty = int(initial_qty * random.uniform(0.95, 1.0)) # 95-100% received purchase_receives.append({ "purchase_order_ref_no": po_ref, "receive_order_ref_no": pr_ref, "product_name": product["product_name"], "vendor_name": vendor, "location_name": location, "receive_order_date": receive_date.strftime("%Y-%m-%d"), "order_qty": initial_qty, "receive_qty": receive_qty, "unit_cost": unit_cost }) inventory[key]["on_hand"] = receive_qty inventory[key]["total_received"] = receive_qty inventory[key]["last_receive_date"] = receive_date inventory[key]["unit_cost"] = unit_cost inventory[key]["vendor"] = vendor inventory[key]["behavior"] = pl["behavior"] inventory[key]["target_ads"] = pl["target_ads"] po_counter += 1 pr_counter += 1 # Phase 2: Simulate day-by-day sales and replenishment print("Phase 2: Simulating daily sales and replenishment...") current_date = start_date + datetime.timedelta(days=7) # Start sales after initial stock days_simulated = 0 while current_date <= today and days_simulated < config['days']: is_weekend = current_date.weekday() >= 5 volume_factor = 1.3 if is_weekend else 1.0 for pl in product_locations: product = pl["product"] location = pl["location"] key = (product["product_name"], location) inv = inventory[key] behavior = pl["behavior"] target_ads = pl["target_ads"] # Skip dead stock (no sales) if behavior == "dead": continue # Calculate today's demand if behavior == "slow": # Slow movers: sell occasionally if random.random() < 0.1: # 10% chance of sale per day demand = random.randint(1, 3) else: demand = 0 else: # Normal products: demand based on ADS with variance base_demand = target_ads * volume_factor demand = int(random.gauss(base_demand, base_demand * 0.3)) demand = max(0, demand) if demand > 0 and inv["on_hand"] >= demand: # Create sales order so_ref = f"SO-{current_date.strftime('%Y%m%d')}-{so_counter:04d}" unit_price = round(inv.get("unit_cost", 50) * random.uniform(1.3, 2.0), 2) sales_orders.append({ "product_name": product["product_name"], "sales_order_ref_no": so_ref, "location_name": location, "order_date": current_date.strftime("%Y-%m-%d"), "quantity": demand, "unit_price": unit_price }) inv["on_hand"] -= demand inv["total_sold"] += demand so_counter += 1 # Check if reorder needed (for normal products) if behavior == "normal" and inv["on_hand"] < target_ads * 14: # Below 2 weeks stock # Create replenishment order reorder_qty = int(target_ads * 30) # Order 30 days worth reorder_qty = max(reorder_qty, 20) vendor = inv.get("vendor", random.choice(pl["vendors"])) unit_cost = inv.get("unit_cost", round(random.uniform(5.0, 300.0), 2)) po_ref = f"PO-{current_date.strftime('%Y%m%d')}-{po_counter:04d}" purchase_orders.append({ "product_name": product["product_name"], "purchase_order_ref_no": po_ref, "vendor_name": vendor, "location_name": location, "purchase_order_date": current_date.strftime("%Y-%m-%d"), "order_qty": reorder_qty, "unit_cost": unit_cost }) # Receive 3-7 days later receive_date = current_date + datetime.timedelta(days=random.randint(3, 7)) if receive_date > today: receive_date = today pr_ref = f"PR-{receive_date.strftime('%Y%m%d')}-{pr_counter:04d}" receive_qty = int(reorder_qty * random.uniform(0.95, 1.0)) purchase_receives.append({ "purchase_order_ref_no": po_ref, "receive_order_ref_no": pr_ref, "product_name": product["product_name"], "vendor_name": vendor, "location_name": location, "receive_order_date": receive_date.strftime("%Y-%m-%d"), "order_qty": reorder_qty, "receive_qty": receive_qty, "unit_cost": unit_cost }) inv["on_hand"] += receive_qty inv["total_received"] += receive_qty po_counter += 1 pr_counter += 1 current_date += datetime.timedelta(days=1) days_simulated += 1 if days_simulated % 30 == 0: print(f" Processed {days_simulated} days...") # Phase 3: Generate returns (small percentage) print("Phase 3: Generating returns...") # Sales returns: 5-8% of sales orders num_sales_returns = max(1, int(len(sales_orders) * random.uniform(0.05, 0.08))) if sales_orders: orders_to_return = random.sample(sales_orders, min(num_sales_returns, len(sales_orders))) for so in orders_to_return: order_date = datetime.datetime.strptime(so['order_date'], "%Y-%m-%d") return_date = order_date + datetime.timedelta(days=random.randint(1, 14)) if return_date > today: return_date = today return_qty = random.randint(1, so['quantity']) sr_ref = f"SR-{return_date.strftime('%Y%m%d')}-{sr_counter:04d}" sales_returns.append({ "sales_order_ref_no": so['sales_order_ref_no'], "return_order_ref_no": sr_ref, "product_name": so['product_name'], "location_name": so['location_name'], "return_date": return_date.strftime("%Y-%m-%d"), "quantity": return_qty }) sr_counter += 1 # Purchase returns: 2-5% of purchase receives num_purchase_returns = max(1, int(len(purchase_receives) * random.uniform(0.02, 0.05))) if purchase_receives: receives_to_return = random.sample(purchase_receives, min(num_purchase_returns, len(purchase_receives))) for pr in receives_to_return: receive_date = datetime.datetime.strptime(pr['receive_order_date'], "%Y-%m-%d") return_date = receive_date + datetime.timedelta(days=random.randint(1, 7)) if return_date > today: return_date = today return_qty = random.randint(1, max(1, int(pr['receive_qty'] * 0.2))) vr_ref = f"VR-{return_date.strftime('%Y%m%d')}-{vr_counter:04d}" purchase_returns.append({ "receive_order_ref_no": pr['receive_order_ref_no'], "return_order_ref_no": vr_ref, "product_name": pr['product_name'], "vendor_name": pr['vendor_name'], "location_name": pr['location_name'], "return_date": return_date.strftime("%Y-%m-%d"), "order_qty": pr['order_qty'], "return_qty": return_qty }) vr_counter += 1 # Phase 4: Generate stock transfers (10-15% of products) print("Phase 4: Generating stock transfers...") # Only transfer products that exist in multiple locations multi_loc_products = [p for p in products if len(p['location_name'].split("|")) >= 2] num_transfers = max(1, int(len(multi_loc_products) * random.uniform(0.10, 0.15))) if multi_loc_products: products_to_transfer = random.sample(multi_loc_products, min(num_transfers, len(multi_loc_products))) for product in products_to_transfer: prod_locs = product['location_name'].split("|") from_loc, to_loc = random.sample(prod_locs, 2) transfer_date = start_date + datetime.timedelta(days=random.randint(10, config['days'] - 1)) if transfer_date > today: transfer_date = today transfer_qty = random.randint(5, 30) st_ref = f"ST-{transfer_date.strftime('%Y%m%d')}-{st_counter:04d}" stock_transfers.append({ "product_name": product['product_name'], "from_location_name": from_loc, "to_location_name": to_loc, "quantity": transfer_qty, "transfer_date": transfer_date.strftime("%Y-%m-%d"), "transfer_ref_no": st_ref }) st_counter += 1 # Print inventory summary print("\nInventory Summary:") total_on_hand = sum(inv["on_hand"] for inv in inventory.values()) total_received = sum(inv["total_received"] for inv in inventory.values()) total_sold = sum(inv["total_sold"] for inv in inventory.values()) print(f" - Total Received: {total_received}") print(f" - Total Sold: {total_sold}") print(f" - Net On-Hand: {total_on_hand}") # Calculate DOH stats doh_values = [] for key, inv in inventory.items(): if inv["total_sold"] > 0 and days_simulated > 0: ads = inv["total_sold"] / days_simulated doh = inv["on_hand"] / ads if ads > 0 else 999 doh_values.append(doh) if doh_values: avg_doh = sum(doh_values) / len(doh_values) print(f" - Average DOH: {avg_doh:.1f} days") return (purchase_orders, purchase_receives, sales_orders, sales_returns, purchase_returns, stock_transfers) def write_csv(filename, data, fieldnames=None): """Write data to CSV file""" if not data: print(f" - {filename}: No data to write") return filepath = os.path.join(OUTPUT_DIR, filename) if fieldnames is None: fieldnames = list(data[0].keys()) with open(filepath, 'w', newline='', encoding='utf-8') as f: writer = csv.DictWriter(f, fieldnames=fieldnames) writer.writeheader() writer.writerows(data) print(f" - {filename}: {len(data)} records") def generate_all_csv(): """Main function to generate all CSV files""" config = get_custom_config() start_date = datetime.datetime.now() - datetime.timedelta(days=config['days']) print(f"\n{'=' * 60}") print("GENERATING REALISTIC INVENTORY DATA") print(f"{'=' * 60}") print(f"Products: {config['products']}") print(f"Locations: {config['locations']}") print(f"Vendors: {config['vendors']}") print(f"Date range: {start_date.date()} to {datetime.datetime.now().date()}") print(f"Output directory: {OUTPUT_DIR}") # Select locations and vendors locations = LOCATION_NAMES[:config['locations']] vendors = VENDOR_NAMES[:config['vendors']] # Generate products print("\nGenerating products...") products = generate_products(config, locations, vendors) # Simulate inventory flow (purchase_orders, purchase_receives, sales_orders, sales_returns, purchase_returns, stock_transfers) = simulate_inventory_flow( config, products, locations, vendors, start_date ) # Write CSV files print("\nWriting CSV files...") write_csv("product.csv", products) write_csv("purchase_order.csv", purchase_orders) write_csv("purchase_receive.csv", purchase_receives) write_csv("sales_order.csv", sales_orders) write_csv("sales_return.csv", sales_returns) write_csv("purchase_return.csv", purchase_returns) write_csv("stock_transfer.csv", stock_transfers) # Print summary print(f"\n{'=' * 60}") print("GENERATION COMPLETE") print(f"{'=' * 60}") print(f"\nFiles generated in: {OUTPUT_DIR}") print(f"\nSummary:") print(f" - Products: {len(products):>6}") print(f" - Purchase Orders: {len(purchase_orders):>6}") print(f" - Purchase Receives: {len(purchase_receives):>6}") print(f" - Sales Orders: {len(sales_orders):>6}") print(f" - Sales Returns: {len(sales_returns):>6}") print(f" - Purchase Returns: {len(purchase_returns):>6}") print(f" - Stock Transfers: {len(stock_transfers):>6}") print(f"\nBehavior Distribution:") print(f" - Target: 81% normal, 14% slow, 5% dead") print(f" - DOH targets: Normal ~30d, Slow ~90d, Dead ~180d") print(f"\nDate Range: {start_date.date()} to {datetime.datetime.now().date()}") print(f"{'=' * 60}\n") if __name__ == "__main__": generate_all_csv()