Price Monitoring Competitor Intelligence Workflows

Rule-Based Repricing Logic in Python

You have a product’s cost, its live price, and a snapshot of where competitors sit, and you need a single function that returns what the new price should be, which rule decided it, and why — deterministically, so the same inputs always give the same answer and every decision is explainable. This is a focused, end-to-end recipe for exactly that: define Rule objects, assemble them into an ordered ruleset, and evaluate a product’s state to produce a proposed price with a fired-rule name and a reason string. It sits under automated repricing rule engines, which covers the surrounding architecture; here we write the runnable core. The proposed price this recipe produces is a candidate — before it reaches a storefront it must pass through the clamp layer in price floors and guardrails in repricing.

Prerequisites & Input Contract

Python 3.11+ and the standard library only — no third-party dependencies. The engine evaluates one immutable snapshot per product; freezing it is what guarantees the function is pure and its output reproducible.

from dataclasses import dataclass
from typing import Callable, Optional

@dataclass(frozen=True)
class ProductState:
    sku: str
    current_price: float     # live storefront price
    landed_cost: float       # unit cost + freight + duties + fees
    competitor_min: float    # lowest in-stock competitor offer
    price_index: float       # our price / market reference (1.0 == parity)
    stock: int               # our on-hand units
    fresh: bool = True        # is the competitive snapshot current this cycle?

The competitor_min and price_index are produced upstream by price positioning and competitive benchmarking; landed_cost must be a true landed figure, not a purchase-order line, or the margin floor below is meaningless.

Signature of the reprice functionA ProductState snapshot enters an ordered ruleset. The evaluate function walks the rules and the first one whose condition matches fires, producing an output triple: a proposed price, the name of the rule that fired, and a reason string.ProductStatefrozen snapshotevaluate(state,ruleset)proposed_pricerule_firedreason

Step-by-Step Implementation

Step 1 — Model a rule as a condition and an action

A rule pairs a predicate over the state with an action that builds a decision. Because both are plain callables, the evaluator never needs to know what any rule does.

@dataclass(frozen=True)
class Decision:
    sku: str
    proposed_price: Optional[float]  # None means "hold, no change"
    rule_fired: str
    reason: str

@dataclass(frozen=True)
class Rule:
    name: str
    when: Callable[[ProductState], bool]      # condition
    then: Callable[[ProductState], Decision]   # action

Step 2 — Write the evaluator with first-match short-circuit

The evaluator walks the ordered list and returns the decision from the first rule whose condition is true. Rule order is the priority: a safety rule at the top can never be overridden by an aggressive rule below.

def evaluate(state: ProductState, ruleset: list[Rule]) -> Decision:
    """First matching rule wins; evaluation stops there (short-circuit)."""
    for rule in ruleset:
        if rule.when(state):
            return rule.then(state)
    return Decision(state.sku, None, "no_rule", "no rule matched")

Step 3 — Define the strategy rules in priority order

Three rules cover the common case. The cost-plus floor protects margin and comes first. The competitive rule undercuts the market when we sit above our target band. A final catch-all holds, which makes evaluate total — every product always exits with an explicit decision.

MIN_MARGIN = 0.15       # 15% over landed cost
UNDERCUT = 0.01         # undercut competitor_min by one cent
TARGET_HI = 1.00        # reprice only when index is above parity

def _floor(s: ProductState) -> float:
    return round(s.landed_cost * (1 + MIN_MARGIN), 2)

cost_plus = Rule(
    name="cost_plus_floor",
    # Following the market would breach our margin floor -> hold at floor.
    when=lambda s: s.competitor_min < _floor(s),
    then=lambda s: Decision(s.sku, _floor(s), "cost_plus_floor",
                            f"market {s.competitor_min:.2f} below floor {_floor(s):.2f}"),
)

competitive = Rule(
    name="competitive",
    # Only fire on a fresh snapshot and when we are priced above the band.
    when=lambda s: s.fresh and s.price_index > TARGET_HI,
    then=lambda s: Decision(s.sku, round(s.competitor_min - UNDERCUT, 2),
                            "competitive", "undercut competitor_min by one cent"),
)

hold = Rule(
    name="hold",
    when=lambda s: True,                       # catch-all, always last
    then=lambda s: Decision(s.sku, None, "hold", "within band; no change"),
)

RULESET = [cost_plus, competitive, hold]

Step 4 — Orchestrate and suppress no-op reprices

The orchestrator runs the evaluator and then collapses a proposal that equals the live price into a hold, so the engine never emits a change worth zero cents of movement — the primary source of self-inflicted churn.

def reprice(state: ProductState, ruleset=RULESET, eps: float = 0.005) -> dict:
    d = evaluate(state, ruleset)
    if d.proposed_price is not None and abs(d.proposed_price - state.current_price) <= eps:
        d = Decision(state.sku, None, "hold", f"proposal == current ({state.current_price:.2f})")
    return {"sku": d.sku, "proposed_price": d.proposed_price,
            "rule_fired": d.rule_fired, "reason": d.reason}


if __name__ == "__main__":
    # We sit above the market (index 1.06) with healthy margin -> undercut.
    s = ProductState("44182", current_price=21.99, landed_cost=14.20,
                     competitor_min=19.49, price_index=1.06, stock=40)
    print(reprice(s))

Running the module prints a single, fully-explained decision:

# {'sku': '44182', 'proposed_price': 19.48, 'rule_fired': 'competitive',
#  'reason': 'undercut competitor_min by one cent'}

Change the inputs so the market sits below the margin floor (competitor_min=15.90) and the same function short-circuits on the cost-plus rule instead, returning proposed_price: 16.33 with rule_fired: 'cost_plus_floor' — the competitive rule never runs.

Verification & Testing

Because the engine is a pure function, tests are exhaustive over the branches rather than statistical. Assert one state per rule plus the idempotency and freshness guards.

def test_reprice():
    base = dict(current_price=21.99, landed_cost=14.20, stock=40)

    # Competitive rule fires when index is above the band on fresh data.
    r = reprice(ProductState("a", competitor_min=19.49, price_index=1.06, **base))
    assert r["rule_fired"] == "competitive" and r["proposed_price"] == 19.48

    # Cost-plus wins by priority when the market is below the floor.
    r = reprice(ProductState("b", competitor_min=15.90, price_index=0.90, **base))
    assert r["rule_fired"] == "cost_plus_floor" and r["proposed_price"] == 16.33

    # Stale snapshot: competitive rule is gated off, engine holds.
    r = reprice(ProductState("c", competitor_min=19.49, price_index=1.06, fresh=False, **base))
    assert r["rule_fired"] == "hold"

    # Idempotency: re-evaluating the proposed price yields no change.
    settled = ProductState("d", current_price=19.48, landed_cost=14.20,
                           competitor_min=19.49, price_index=1.00, stock=40)
    assert reprice(settled)["proposed_price"] is None
    print("all rules covered")

test_reprice()   # -> all rules covered

The idempotency assertion is the one that matters most in production: feed the engine the price it just proposed and it must settle to a hold, not cut again. A ruleset that fails this test will spiral prices downward one cent per cycle.

Edge Cases & Gotchas

  • Self-referential competitive spirals. If a competitive rule targeted current_price - delta instead of an external anchor, each cycle would treat its own last move as the market and keep cutting. Always anchor the action to an external value (competitor_min), and keep the no-op suppression in reprice as the second line of defense.
  • Rule overlap and shadowing. A broad early condition can shadow a later rule so it never fires. Because short-circuit means only the first match runs, order the ruleset from most specific and most protective to most general, and lint offline by asserting each rule fires at least once across a sample of real states.
  • Missing catch-all. Drop the hold rule and evaluate falls through to the no_rule sentinel — technically total, but it signals a misconfigured ruleset. Treat any no_rule result in production as an alert, not a silent hold.
  • Stale competitive data. A scrape that failed silently leaves competitor_min stale; acting on it reprices to a phantom market. The fresh flag gates the competitive rule off, so the engine degrades to holding rather than to a wrong price.

Performance Notes

Evaluation is O(r) in the number of rules per product, and r is a handful, so the engine itself is never the bottleneck — a single Python worker evaluates well over 100k states per second because each rule is a couple of comparisons and no allocation of note. The cost lives entirely in assembling ProductState (the upstream joins against the competitive index and cost tables), so batch those, then evaluate the frozen snapshots in a tight loop or a vectorized pass. Keep evaluation synchronous and deterministic; push the slow work — I/O to fetch inputs and dispatch approved changes — onto async workers so a slow marketplace API never blocks the decision core.

Frequently Asked Questions

Should the rule engine write the price directly to the marketplace? No. The engine returns a proposed price only. Dispatching is a separate step that runs after the proposal passes the guardrail clamp — hard floors, ceilings, max-step, and cool-down — so strategy and inviolable limits stay decoupled.

How do I stop the engine from cutting a price every single cycle? Anchor competitive actions to an external value such as the competitor’s minimum rather than to your own current price, and suppress any proposal that equals the live price within a small epsilon. Together these make the engine idempotent, so an unchanged market yields an unchanged decision.

What happens when no rule matches a product? End every ruleset with a catch-all hold rule so evaluate always returns an explicit decision. If you ever see the no_rule sentinel in production, treat it as a configuration alert rather than a silent hold, because it means the catch-all is missing or misordered.