pylitmus 1.2.0
A high-performance rules engine for Python - evaluate data against configurable rules and get clear verdicts
License: MIT
Requires Python: >=3.11
Keywords: business-rules, decision-engine, fraud-detection, litmus-test, risk-assessment, rules-engine, scoring-engine

# pylitmus

A high-performance rules engine for Python. Like a litmus test for your data - evaluate against configurable rules and get clear verdicts.

[![PyPI version](https://badge.fury.io/py/pylitmus.svg)](https://badge.fury.io/py/pylitmus)
[![Python](https://img.shields.io/pypi/pyversions/pylitmus.svg)](https://pypi.org/project/pylitmus/)
[![License](https://img.shields.io/badge/license-MIT-blue.svg)](LICENSE)
[![Tests](https://github.com/org/pylitmus/workflows/Tests/badge.svg)](https://github.com/org/pylitmus/actions)
[![Coverage](https://codecov.io/gh/org/pylitmus/branch/main/graph/badge.svg)](https://codecov.io/gh/org/pylitmus)

## Features

- **YAML/JSON rule definitions** - Business-friendly rule configuration
- **Hot-reload** - Rules can be updated without restart
- **Multiple storage backends** - Memory, database, file
- **Caching** - Redis and in-memory caching support
- **Multiple scoring strategies** - Sum, weighted, max
- **Flask integration** - Easy integration with Flask apps
- **18 built-in operators** - Comparison, collection, string, null, temporal
- **Extensible** - Custom evaluators and strategies

## Installation

```bash
pip install pylitmus

# With database support
pip install pylitmus[database]

# With Redis caching
pip install pylitmus[redis]

# With Flask integration
pip install pylitmus[flask]

# All extras
pip install pylitmus[all]
```

## Quick Start

```python
from pylitmus import create_engine, Rule, Severity

# Create engine with inline rules
engine = create_engine(rules=[
    Rule(
        code='AMT_001',
        name='High Amount',
        description='Flag high amounts',
        category='AMOUNT',
        severity=Severity.HIGH,
        score=60,
        enabled=True,
        conditions={'field': 'amount', 'operator': 'greater_than', 'value': 5000}
    )
])

# Evaluate data
result = engine.evaluate({'amount': 6000})

print(f"Score: {result.total_score}")      # 60
print(f"Decision: {result.decision}")       # FLAG
print(f"Triggered: {[r.rule_code for r in result.triggered_rules]}")  # ['AMT_001']
```

## YAML Rules

Define rules in YAML files for easy management:

```yaml
# rules.yaml
rules:
  - code: "AMT_001"
    name: "High Amount"
    description: "Flag transactions over $5000"
    category: "AMOUNT"
    severity: "HIGH"
    score: 60
    enabled: true
    conditions:
      field: "amount"
      operator: "greater_than"
      value: 5000

  - code: "RISK_001"
    name: "High Risk Country"
    description: "Flag transactions from high-risk countries"
    category: "RISK"
    severity: "CRITICAL"
    score: 80
    enabled: true
    conditions:
      field: "country"
      operator: "in"
      value: ["XX", "YY", "ZZ"]
```

Load rules from file:

```python
engine = create_engine(
    storage_backend='file',
    rules_file='rules.yaml'
)
```

## Composite Conditions

Combine conditions with AND/OR logic:

```yaml
conditions:
  all:  # AND
    - field: "amount"
      operator: "greater_than"
      value: 1000
    - any:  # OR
        - field: "is_new_customer"
          operator: "equals"
          value: true
        - field: "country"
          operator: "in"
          value: ["NG", "KE", "GH"]
```

Or use the alternative format:

```yaml
conditions:
  type: "AND"
  conditions:
    - field: "amount"
      operator: "greater_than"
      value: 1000
    - field: "is_international"
      operator: "equals"
      value: true
```

## Available Operators

| Category | Operators |
|----------|-----------|
| **Comparison** | `equals`, `not_equals`, `greater_than`, `greater_than_or_equal`, `less_than`, `less_than_or_equal`, `between` |
| **Collection** | `in`, `not_in`, `contains`, `not_contains` |
| **String** | `starts_with`, `ends_with`, `matches_regex` |
| **Null** | `is_null`, `is_not_null` |
| **Temporal** | `within_days`, `before`, `after` |

## Scoring Strategies

### Sum Strategy (Default)
Adds up all triggered rule scores, capped at 100.

```python
engine = create_engine(scoring_strategy='sum')
```

### Weighted Strategy
Uses severity-based weights (LOW=1, MEDIUM=2, HIGH=3, CRITICAL=4).

```python
engine = create_engine(scoring_strategy='weighted')
```

### Max Strategy
Takes the highest score from triggered rules.

```python
engine = create_engine(scoring_strategy='max')
```

## Decision Thresholds

Customize decision boundaries:

```python
engine = create_engine(
    decision_thresholds={
        'approve': 30,  # Score < 30 = APPROVE
        'review': 70    # Score 30-70 = REVIEW, >= 70 = FLAG
    }
)
```

## Storage Backends

### In-Memory
```python
engine = create_engine(storage_backend='memory', rules=[...])
```

### File-Based
```python
engine = create_engine(
    storage_backend='file',
    rules_file='rules.yaml'  # or rules.json
)
```

### Database
```python
engine = create_engine(
    storage_backend='database',
    database_url='postgresql://localhost/mydb'
)
```

## Caching

### Memory Cache
```python
engine = create_engine(
    cache_backend='memory',
    cache_ttl=300  # 5 minutes
)
```

### Redis Cache
```python
engine = create_engine(
    cache_backend='redis',
    cache_url='redis://localhost:6379/0',
    cache_ttl=600
)
```

### No Cache
```python
engine = create_engine(cache_backend='none')
```

## Flask Integration

```python
from flask import Flask
from pylitmus.integrations.flask import CmapRulesEngine, get_engine

app = Flask(__name__)
app.config['CMAP_RULES_FILE'] = 'rules.yaml'

rules_engine = CmapRulesEngine(app)

@app.route('/evaluate', methods=['POST'])
def evaluate():
    data = request.json
    engine = get_engine()
    result = engine.evaluate(data)
    return {
        'score': result.total_score,
        'decision': result.decision,
        'triggered_rules': [r.rule_code for r in result.triggered_rules]
    }
```

## Nested Field Access

Access nested data using dot notation:

```python
data = {
    'transaction': {
        'amount': 6000,
        'merchant': {
            'category': 'electronics'
        }
    }
}

# Rule condition
conditions:
  field: "transaction.merchant.category"
  operator: "equals"
  value: "electronics"
```

## Pattern Matching

Advanced pattern matching capabilities:

```python
from pylitmus import EnhancedPatternEngine

pattern_engine = EnhancedPatternEngine()

# Regex matching
pattern_engine.add_pattern('email', r'^[\w.-]+@[\w.-]+\.\w+$', 'regex')

# Fuzzy matching
pattern_engine.add_pattern('name', 'John Smith', 'fuzzy', threshold=0.8)

# Range matching
pattern_engine.add_pattern('age', {'min': 18, 'max': 65}, 'range')

# Check matches
result = pattern_engine.match_all({
    'email': 'user@example.com',
    'name': 'Jon Smith',
    'age': 25
})
```

## API Reference

### create_engine()

```python
def create_engine(
    storage_backend: str = 'memory',
    database_url: str = None,
    rules_file: str = None,
    rules: List[Rule] = None,
    repository: RuleRepository = None,
    cache_backend: str = 'memory',
    cache_url: str = None,
    cache_ttl: int = 300,
    scoring_strategy: str = 'sum',
    decision_thresholds: Dict[str, int] = None,
) -> RuleEngine
```

### RuleEngine.evaluate()

```python
def evaluate(
    self,
    data: Dict[str, Any],
    context: Dict[str, Any] = None,
    filters: Dict[str, Any] = None
) -> AssessmentResult
```

### AssessmentResult

```python
@dataclass
class AssessmentResult:
    total_score: int           # Total calculated score
    decision: str              # APPROVE, REVIEW, or FLAG
    triggered_rules: List[RuleResult]  # Rules that matched
    processing_time_ms: float  # Processing time in ms
```

## Full Example

```python
from pylitmus import (
    create_engine,
    Rule,
    Severity,
    InMemoryRuleRepository,
    WeightedStrategy,
)

# Define rules
rules = [
    Rule(
        code='AMT_HIGH',
        name='High Amount',
        description='Flag high-value transactions',
        category='AMOUNT',
        severity=Severity.HIGH,
        score=60,
        enabled=True,
        conditions={'field': 'amount', 'operator': 'greater_than', 'value': 5000}
    ),
    Rule(
        code='NEW_CUSTOMER',
        name='New Customer',
        description='Flag new customer transactions',
        category='CUSTOMER',
        severity=Severity.MEDIUM,
        score=30,
        enabled=True,
        conditions={'field': 'is_new_customer', 'operator': 'equals', 'value': True}
    ),
    Rule(
        code='INTL_TXN',
        name='International Transaction',
        description='Flag international transactions',
        category='GEOGRAPHY',
        severity=Severity.LOW,
        score=20,
        enabled=True,
        conditions={'field': 'is_international', 'operator': 'equals', 'value': True}
    ),
]

# Create engine with weighted scoring
engine = create_engine(
    rules=rules,
    scoring_strategy='weighted',
    decision_thresholds={'approve': 25, 'review': 60}
)

# Evaluate transaction
result = engine.evaluate({
    'amount': 6000,
    'is_new_customer': True,
    'is_international': False
})

print(f"Total Score: {result.total_score}")
print(f"Decision: {result.decision}")
print(f"Triggered Rules: {[r.rule_code for r in result.triggered_rules]}")
print(f"Processing Time: {result.processing_time_ms:.2f}ms")
```

## Documentation

- [Quick Start Guide](docs/quickstart.md)
- [Rule Format](docs/rules-format.md)
- [API Reference](docs/api-reference.md)
- [Flask Integration](docs/flask-integration.md)
- [Examples](examples/)

## License

MIT License - see [LICENSE](LICENSE) for details.