Skip to main content

Overview

Stock Movements track every inventory transaction in the system. Each movement records quantity changes, pricing, costs, and links to batches for complete audit trails.
All inventory changes are recorded as movements, creating a complete history of stock flow through your business.

Movement Types

The system supports three primary movement types:

ENTRY

Inventory additions from purchases or production

EXIT

Inventory reductions from sales or consumption

ADJUSTMENT

Manual corrections, write-offs, or inventory counts

Movement Data Model

class MovementType(str, enum.Enum):
    ENTRY = "ENTRY"
    EXIT = "EXIT"
    ADJUSTMENT = "ADJUSTMENT"

class Movement(Base, AuditableEntity):
    __tablename__ = "movements"

    id = Column(String(50), primary_key=True, index=True)
    product_id = Column(String(50), ForeignKey("products.id"), nullable=False)
    supplier_id = Column(String(50), ForeignKey("suppliers.id"), nullable=True)
    customer_id = Column(String(50), ForeignKey("customers.id"), nullable=True)
    type = Column(String(50), nullable=False)  # MovementType values
    quantity = Column(Integer, nullable=False)
    unit_price = Column(Float, nullable=True)  # Price applied to transaction
    total_price = Column(Float, nullable=True)  # quantity * unit_price
    total_cost = Column(Float, nullable=True)  # FIFO calculated cost for EXIT
    reference_id = Column(String(50), nullable=True)  # Order ID or Batch ID
    notes = Column(String(500), nullable=True)

Key Fields

type
enum
required
Movement type: ENTRY, EXIT, or ADJUSTMENT
quantity
integer
required
Number of units moved (positive integer)
unit_price
float
Selling price per unit (for EXIT movements)
total_price
float
Total revenue: quantity × unit_price (for EXIT movements)
total_cost
float
FIFO-calculated cost of goods sold (for EXIT movements)
reference_id
string
Links to batch IDs (comma-separated for multi-batch exits) or order IDs
supplier_id
string
Supplier reference for ENTRY movements
customer_id
string
Customer reference for EXIT movements (sales)

ENTRY Movements

Entry movements record inventory additions and create new batches.

Creating an Entry

def register_entry(self, product_id: str, quantity: int, unit_cost: float, 
                  supplier_id: Optional[str] = None, 
                  expiration_date: Optional[datetime] = None) -> Tuple[Batch, Movement]:
    if quantity <= 0:
        raise HTTPException(status_code=400, detail="Quantity must be greater than 0")
    
    product = self.product_repo.get_by_id(product_id)
    if not product:
        raise HTTPException(status_code=404, detail="Product not found")

    # Create Batch
    batch_id = str(uuid.uuid4())
    new_batch = Batch(
        id=batch_id,
        product_id=product_id,
        initial_quantity=quantity,
        available_quantity=quantity,
        unit_cost=unit_cost,
        purchase_date=datetime.now(timezone.utc),
        expiration_date=expiration_date,
        supplier_id=supplier_id
    )
    self.repo.create_batch(new_batch)

    # Create Entry Movement
    mov_id = str(uuid.uuid4())
    new_movement = Movement(
        id=mov_id,
        product_id=product_id,
        type=MovementType.ENTRY,
        quantity=quantity,
        unit_price=unit_cost,
        total_price=unit_cost * quantity,
        reference_id=batch_id,
        notes="Purchase Entry"
    )
    self.repo.create_movement(new_movement)
    
    return new_batch, new_movement
1

Validate Input

Ensure quantity > 0 and product exists
2

Create Batch

Generate new batch with quantity and cost details
3

Create Movement

Record the ENTRY movement linked to the new batch
4

Return Both

Return batch and movement for confirmation

Entry Movement Attributes

FieldValue for ENTRY
typeENTRY
quantityUnits purchased
unit_priceCost per unit
total_priceTotal purchase cost
total_costnull (not applicable)
reference_idNew batch ID
supplier_idSupplier who provided goods

EXIT Movements

Exit movements record sales or consumption using automatic FIFO batch deduction.

Registering a Sale

@router.route('/sale', methods=['POST'])
@require_role('admin', 'gestor')
def register_sale():
    data = request.get_json()
    
    unit_price = data.get('unit_price')
    if unit_price is not None:
        unit_price = float(unit_price)

    mov, cost = stock_svc.register_exit(
        product_id=data.get('product_id'),
        quantity=int(data.get('quantity')),
        unit_price=unit_price,
        notes=data.get('notes', 'Sale')
    )
    
    return jsonify({
        "movement_id": mov.id,
        "product_id": mov.product_id,
        "quantity": mov.quantity,
        "unit_price": mov.unit_price,
        "total_price": mov.total_price,
        "total_cost": cost
    }), 200
{
  "product_id": "uuid-product",
  "quantity": 50,
  "unit_price": 18.50,
  "notes": "Sale to customer ABC"
}

FIFO Exit Logic

def register_exit(self, product_id: str, quantity: int, 
                 unit_price: Optional[float] = None, 
                 notes: str = "") -> Tuple[Movement, float]:
    # Use suggested price if no price provided
    applied_unit_price = unit_price if unit_price is not None else product.suggested_price

    # Get active batches sorted by expiration/purchase date
    batches = self.repo.get_batches_by_product(product_id, active_only=True)
    batches.sort(key=lambda b: (
        b.expiration_date if b.expiration_date else datetime.max,
        b.purchase_date if b.purchase_date else datetime.max
    ))

    # Check sufficient stock
    total_available = sum(b.available_quantity for b in batches)
    if total_available < quantity:
        raise HTTPException(status_code=400, detail="Insufficient stock")

    # Deduct from batches using FIFO
    remaining_to_deduct = quantity
    total_cost = 0.0
    affected_batches = []

    for batch in batches:
        if remaining_to_deduct <= 0:
            break
        
        deduct_amount = min(batch.available_quantity, remaining_to_deduct)
        batch.available_quantity -= deduct_amount
        remaining_to_deduct -= deduct_amount
        total_cost += deduct_amount * batch.unit_cost
        affected_batches.append(batch.id)

    # Create EXIT Movement
    new_movement = Movement(
        id=str(uuid.uuid4()),
        product_id=product_id,
        type=MovementType.EXIT,
        quantity=quantity,
        unit_price=applied_unit_price,
        total_price=applied_unit_price * quantity,
        total_cost=total_cost,
        reference_id=",".join(affected_batches),
        notes=notes
    )
The reference_id for EXIT movements contains comma-separated batch IDs, showing exactly which batches were used to fulfill the sale.

Exit Movement Attributes

FieldValue for EXIT
typeEXIT
quantityUnits sold
unit_priceSelling price per unit
total_priceRevenue: quantity × unit_price
total_costCOGS from FIFO calculation
reference_idAffected batch IDs (comma-separated)
customer_idCustomer who purchased (optional)

Profit Calculation

Each EXIT movement enables profit analysis:
const profit = movement.total_price - movement.total_cost;
const margin = (profit / movement.total_price) * 100;

console.log(`Revenue: Bs ${movement.total_price.toFixed(2)}`);
console.log(`COGS: Bs ${movement.total_cost.toFixed(2)}`);
console.log(`Profit: Bs ${profit.toFixed(2)}`);
console.log(`Margin: ${margin.toFixed(1)}%`);

ADJUSTMENT Movements

Adjustments handle special cases like inventory corrections, write-offs, or physical count discrepancies.
Adjustments should be used sparingly and always include detailed notes explaining the reason for the adjustment.

Common Adjustment Scenarios

When actual inventory doesn’t match system records after a physical count.
Writing off inventory that was damaged or defective.
Removing expired perishable items from inventory.
Recording inventory losses due to theft or unexplained disappearance.
Returning defective items to suppliers.

Viewing Movement History

Query movements by product to see complete transaction history:
@router.route('/product/<product_id>', methods=['GET'])
@require_role('admin', 'gestor', 'consultor')
def get_movements_by_product(product_id):
    movements = repo.get_movements_by_product(product_id)
    result = []
    for m in movements:
        result.append({
            "id": m.id,
            "product_id": m.product_id,
            "type": str(m.type),
            "quantity": m.quantity,
            "unit_price": m.unit_price,
            "total_price": m.total_price,
            "total_cost": m.total_cost,
            "reference_id": m.reference_id,
            "notes": m.notes,
            "created_at": m.created_at.isoformat(),
            "created_by": m.created_by
        })
    return jsonify(result), 200

Movement Display

{recentMovements.map(movement => (
    <tr key={movement.id}>
        <td>{new Date(movement.created_at).toLocaleString('es-ES')}</td>
        <td>{movement.product_name}</td>
        <td>
            <span className={`badge ${
                movement.type === 'ENTRY'
                    ? 'bg-green-500/10 text-green-400'
                    : 'bg-red-500/10 text-red-400'
            }`}>
                {movement.type}
            </span>
        </td>
        <td>{movement.quantity}</td>
        <td>Bs {movement.total_price?.toFixed(2)}</td>
        <td>{movement.created_by || 'Sistema'}</td>
    </tr>
))}

Advanced Movement Queries

The system provides sophisticated filtering for movement analysis:
@router.route('/movements/search', methods=['GET'])
@require_role('admin', 'gestor', 'consultor')
def search_movements():
    # Parse parameters
    start_date = request.args.get('start_date')
    end_date = request.args.get('end_date')
    product_id = request.args.get('product_id')
    movement_type = request.args.get('movement_type')
    user_id = request.args.get('user_id')
    skip = int(request.args.get('skip', 0))
    limit = int(request.args.get('limit', 100))
    
    movements = repo.search_movements(
        start_date=start_date,
        end_date=end_date,
        product_id=product_id,
        movement_type=movement_type,
        user_id=user_id,
        skip=skip,
        limit=limit
    )
    
    return jsonify({"movements": movements}), 200

Query Filters

start_date
date
Filter movements after this date (YYYY-MM-DD)
end_date
date
Filter movements before this date (YYYY-MM-DD)
product_id
string
Filter by specific product
movement_type
enum
Filter by type: ENTRY, EXIT, or ADJUSTMENT
user_id
string
Filter by user who created the movement
skip
integer
default:"0"
Pagination offset
limit
integer
default:"100"
Maximum results to return

Audit Trail

All movements inherit from AuditableEntity, automatically tracking:

Creation Time

created_at timestamp when the movement was recorded

Created By

created_by user ID who initiated the movement

Last Modified

updated_at timestamp of last modification

Modified By

updated_by user ID who last modified the record
Movements are generally immutable once created. Corrections should be made through ADJUSTMENT movements rather than modifying historical records.

Best Practices

Include descriptive notes for every movement, especially adjustments, to maintain clear audit trails.
For EXIT movements, use actual selling prices to enable accurate profit analysis.
The system checks stock availability, but verify expectations before large sales.
Adjustments should always have detailed notes explaining why the correction was necessary.

Movement Analytics

Movements power various reports and analytics:
  • Inventory Rotation: Analyze entry vs. exit patterns
  • Profit Margins: Compare total_price vs. total_cost for EXIT movements
  • User Activity: Track which users are creating the most movements
  • Product Velocity: Identify fast vs. slow-moving products
See Inventory Reports for detailed analytics.

Batch Tracking

Understand FIFO batch management

Inventory Reports

Analyze movement data and trends

API Reference

Complete API documentation

Build docs developers (and LLMs) love