/2

Good. This is the right abstraction. The inventory simulator should not know where demand comes from. It should only receive tasks and process them.

The structure becomes:

inventory_sim/
│
├── materials.csv              # Master data
├── demand_generator.py        # Creates tasks
├── simulation.py              # Inventory logic only
├── models.py                  # Data objects
└── main.py                    # Runs simulation

The flow is now:

Demand Generator
        |
        v
   Task / Auftrag
        |
        v
Inventory Simulation
        |
        +--> Consumption CSV
        |
        +--> Purchase CSV
        |
        +--> Inventory Snapshot


---

models.py

Add a proper task model.

from dataclasses import dataclass


@dataclass
class Material:
    material_id: str
    description: str
    stock: int
    unit_price: float
    min_stock: int
    max_stock: int


@dataclass
class TaskLine:
    material_id: str
    quantity: int


@dataclass
class Task:
    task_id: str
    day: int
    lines: list[TaskLine]


---

demand_generator.py

This is now responsible only for creating demand.

Later this can be replaced with SAP data.

import random

from models import Task, TaskLine


class DemandGenerator:

    def __init__(self, material_ids):

        self.material_ids = material_ids

        self.task_counter = 0


    def generate_day(self, day):

        tasks = []

        number_of_tasks = random.randint(0, 5)

        for _ in range(number_of_tasks):

            self.task_counter += 1

            lines = []

            number_of_materials = random.randint(1, 4)

            selected = random.sample(
                self.material_ids,
                number_of_materials
            )

            for material_id in selected:

                lines.append(
                    TaskLine(
                        material_id,
                        random.randint(1, 8)
                    )
                )


            tasks.append(
                Task(
                    task_id=f"TASK-{self.task_counter:06}",
                    day=day,
                    lines=lines
                )
            )

        return tasks

Example generated demand:

Day 15

TASK-000231

M001 Bearing     x3
M004 Shaft       x1


TASK-000232

M002 Bolt        x15
M003 Nut         x8


---

simulation.py

Remove all random generation.

The simulator only receives tasks.

Replace the old task generation part with:

def process_day(self, day, tasks):

    self.day = day

    for task in tasks:

        self.process_task(task)

And modify process_task:

def process_task(self, task):

    for line in task.lines:

        material = self.materials[line.material_id]


        consumed = min(
            line.quantity,
            material.stock
        )


        material.stock -= consumed


        self.consumption_history.append(
            {
                "day": task.day,
                "task_id": task.task_id,
                "material": material.material_id,
                "quantity": consumed
            }
        )


        self.check_replenishment(material)


---

Add a separate purchasing function:

def check_replenishment(self, material):

    if material.stock <= material.min_stock:


        quantity = (
            material.max_stock
            - material.stock
        )


        cost = quantity * material.unit_price


        if self.money >= cost:


            self.money -= cost

            material.stock += quantity


            self.purchase_history.append(
                {
                    "day": self.day,
                    "material": material.material_id,
                    "quantity": quantity,
                    "cost": cost
                }
            )

Now purchasing is isolated.


---

main.py

Now the orchestration is very clear.

from models import Material
from simulation import InventorySimulation
from demand_generator import DemandGenerator
import csv



def load_materials(filename):

    materials=[]

    with open(filename) as f:

        reader=csv.DictReader(f)

        for row in reader:

            materials.append(
                Material(
                    row["material_id"],
                    row["description"],
                    int(row["stock"]),
                    float(row["unit_price"]),
                    int(row["min_stock"]),
                    int(row["max_stock"])
                )
            )

    return materials



materials = load_materials(
    "materials.csv"
)



simulation = InventorySimulation(
    materials,
    starting_money=50000
)


demand = DemandGenerator(
    [
        m.material_id
        for m in materials
    ]
)



simulation_days = 365



for day in range(1, simulation_days + 1):

    tasks = demand.generate_day(day)

    simulation.process_day(
        day,
        tasks
    )



simulation.export_csv()


---

Now the CSV outputs become more realistic

consumption_history.csv

day,task_id,material,quantity
15,TASK-000231,M001,3
15,TASK-000231,M004,1
15,TASK-000232,M002,15

This represents your "System Change" / consumption table.


---

purchase_history.csv

day,material,quantity,cost
15,M002,45,90
23,M001,33,495


---

Future replacement with real data

The nice thing now:

Today:

DemandGenerator
       |
       |
random tasks

Later:

DemandGenerator
       |
       |
SAP CSV import
       |
       |
real Auftrag numbers

The simulator does not change.

Your future SAP export would just need something like:

task_id,day,material_id,quantity
4711001,1,M001,4
4711001,1,M002,10
4711002,1,M003,2

and you replace only demand_generator.py.

This is now much closer to a real spare-parts inventory digital twin architecture.