Hierarchical Cost-Flow Traceback

The problem

It is difficult to see exactly which specific user action or input is causing a sudden spike in costs. Identifying the root cause of high expenses is often buried under complex data.

What it does

It maps specific user inputs directly to the resources they consume to pinpoint the exact source of a cost increase. It traces the path from a single action to the final cost.

Why it matters

It provides a clear link between user behavior and financial expenses.

Validation

It was run in the sandbox and it failed. run output shows an error/traceback — the artifact does NOT run clean.

$ python3 cost_flow_traceback.py
Traceback (most recent call last):
  File "/work/cost_flow_traceback.py", line 74, in <module>
    spikes = analyzer.identify_cost_spikes()
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
AttributeError: 'CostFlowAnalyzer' object has no attribute 'identify_cost_spikes'

No screenshot — there is nothing working to show. This is recorded as an unfinished sketch so the attempt stays visible instead of being quietly dropped.

The code

All of it — 78 lines, one file, standard library only.

# Hierarchical Cost-Flow Traceback Implementation

class StatePath:
    def __init__(self, state, input_source, cost=0):
        self.state = state
        self.input_source = input_source
        self.cost = cost
        self.children = []

    def add_child(self, child):
        self.children.append(child)

    def calculate_cost_flow(self):
        total_cost = self.cost
        for child in self.children:
            total_cost += child.calculate_cost_flow()
        return total_cost


class CostFlowAnalyzer:
    def __init__(self, traces):
        self.traces = traces
        self.state_map = {}

def build_hierarchy(self):
    for trace in self.traces:
        current_node = StatePath(trace['initial_state'], 'root', 0)
        self.state_map[current_node.state] = current_node
        for step in trace['steps']:
            new_node = StatePath(step['state'], step['input'], step['cost'])
            current_node.add_child(new_node)
            current_node = new_node

def identify_cost_spikes(self, threshold=0.8):
    self.build_hierarchy()
    spikes = []
    for root in self.state_map.values():
        total_cost = root.calculate_cost_flow()
        for state in self._dfs(root):
            if state.cost / total_cost > threshold and state.input_source != 'system':
                spikes.append((state, state.cost, state.input_source))
    return spikes

    def _dfs(self, node):
        stack = [node]
        while stack:
            current = stack.pop()
            yield current
            for child in reversed(current.children):
                stack.append(child)


# Example Usage
if __name__ == '__main__':
    # Sample execution traces
    traces = [{
        'initial_state': 'start',
        'steps': [{
            'state': 'A',
            'input': 'user_1',
            'cost': 10
        }, {
            'state': 'B',
            'input': 'user_2',
            'cost': 50  # Simulated cost spike
        }, {
            'state': 'C',
            'input': 'system',
            'cost': 5
        }]
    }]

    analyzer = CostFlowAnalyzer(traces)
    spikes = analyzer.identify_cost_spikes()

    print('Identified cost spikes:')
    for state, cost in spikes:
        print(f"State {state.state} (Input: {state.input_source}) - Cost: {cost}")