Deterministic credit limit decision service for trade credit insurance underwriters
Provides instant APPROVE/REFER/DECLINE decisions based on buyer risk grade and payment history, with full explainability (reason codes + human explanations) and regulatory audit trail.
- ✅ Automated Decisions: Instant credit limit approvals, referrals, and declines
- ✅ Explainable: Every decision includes reason codes and human-readable explanations
- ✅ Auditable: Immutable audit trail in DynamoDB for regulatory compliance
- ✅ Deterministic: Same inputs always produce same decision outcomes
- ✅ Pure Function Logic: Decision engine has zero I/O operations for testability
- ✅ Type-Safe: Full Python type hints with dataclasses throughout
API Gateway HTTP API → Lambda (Python 3.13) → DynamoDB (Audit)
↓
Pure Decision Engine
↓
Stubbed Risk Data (MVP)
Technology Stack:
- AWS Lambda (Python 3.13 runtime)
- API Gateway HTTP API (Lambda proxy integration)
- DynamoDB (audit table: CreditDecisionAudit)
- AWS SAM (Infrastructure as Code)
- pytest (testing framework)
- Python 3.13+
- AWS CLI configured with credentials
- AWS SAM CLI installed
- pytest installed (
pip install pytest pytest-cov)
-
Clone and install dependencies:
git clone <repo-url> cd <repo-name> pip install -r requirements.txt pip install pytest pytest-cov
-
Run unit tests:
# Run all tests pytest # Run with coverage pytest --cov=src/credit_decision --cov-report=html # Run specific test file pytest tests/test_decision_engine.py -v
-
Test locally with SAM CLI:
# Build the SAM application sam build # Start local API Gateway sam local start-api # In another terminal, test the endpoint (see examples below)
-
Build and deploy:
sam build sam deploy --guided
-
Follow the prompts:
- Stack name:
credit-decision-service - AWS Region:
us-east-1(or your preferred region) - Confirm changes before deploy:
Y - Allow SAM CLI IAM role creation:
Y - Disable rollback:
N - Save arguments to configuration file:
Y
- Stack name:
-
Get the API endpoint:
aws cloudformation describe-stacks \ --stack-name credit-decision-service \ --query "Stacks[0].Outputs[?OutputKey=='CreditDecisionApi'].OutputValue" \ --output text
POST /credit-decisions
{
"buyerId": "string (required)",
"policyId": "string (required)",
"requestedLimit": "integer (required, > 0, no decimals)",
"currency": "string (required, 3 uppercase letters)",
"requestId": "string (optional, idempotency key)"
}{
"decisionId": "uuid",
"decision": "APPROVE | REFER | DECLINE",
"approvedLimit": "integer (0 for REFER/DECLINE)",
"currency": "string",
"reasonCodes": ["array of reason codes"],
"explanations": ["array of human-readable explanations"],
"timestamp": "ISO-8601 UTC"
}{
"errorCode": "ERROR_CODE",
"message": "Human-readable error message"
}Request:
curl -X POST http://127.0.0.1:3000/credit-decisions \
-H "Content-Type: application/json" \
-d '{
"buyerId": "BYR-A-CLEAN",
"policyId": "POL-67890",
"requestedLimit": 750000,
"currency": "USD"
}'Response (HTTP 200):
{
"decisionId": "550e8400-e29b-41d4-a716-446655440000",
"decision": "APPROVE",
"approvedLimit": 750000,
"currency": "USD",
"reasonCodes": [],
"explanations": [],
"timestamp": "2025-12-30T10:30:00.000Z"
}Request:
curl -X POST http://127.0.0.1:3000/credit-decisions \
-H "Content-Type: application/json" \
-d '{
"buyerId": "BYR-A-CLEAN",
"policyId": "POL-67890",
"requestedLimit": 1500000,
"currency": "EUR"
}'Response (HTTP 200):
{
"decisionId": "660e9400-e29b-41d4-a716-446655440001",
"decision": "APPROVE",
"approvedLimit": 1000000,
"currency": "EUR",
"reasonCodes": ["LIMIT_CAPPED_BY_GRADE"],
"explanations": ["Requested limit exceeds maximum allowed for buyer's risk grade"],
"timestamp": "2025-12-30T10:31:00.000Z"
}Request:
curl -X POST http://127.0.0.1:3000/credit-decisions \
-H "Content-Type: application/json" \
-d '{
"buyerId": "BYR-B-CLEAN",
"policyId": "POL-12345",
"requestedLimit": 400000,
"currency": "GBP"
}'Response (HTTP 200):
{
"decisionId": "770e9400-e29b-41d4-a716-446655440002",
"decision": "APPROVE",
"approvedLimit": 400000,
"currency": "GBP",
"reasonCodes": [],
"explanations": [],
"timestamp": "2025-12-30T10:32:00.000Z"
}Request:
curl -X POST http://127.0.0.1:3000/credit-decisions \
-H "Content-Type: application/json" \
-d '{
"buyerId": "BYR-C-MEDIUM",
"policyId": "POL-11111",
"requestedLimit": 300000,
"currency": "USD"
}'Response (HTTP 200):
{
"decisionId": "880e9400-e29b-41d4-a716-446655440003",
"decision": "REFER",
"approvedLimit": 0,
"currency": "USD",
"reasonCodes": ["RISK_GRADE_MEDIUM"],
"explanations": ["Buyer risk grade (C) requires manual underwriter review"],
"timestamp": "2025-12-30T10:33:00.000Z"
}Request:
curl -X POST http://127.0.0.1:3000/credit-decisions \
-H "Content-Type: application/json" \
-d '{
"buyerId": "BYR-A-PASTDUE",
"policyId": "POL-22222",
"requestedLimit": 500000,
"currency": "USD"
}'Response (HTTP 200):
{
"decisionId": "990e9400-e29b-41d4-a716-446655440004",
"decision": "REFER",
"approvedLimit": 0,
"currency": "USD",
"reasonCodes": ["PAST_DUE_OVER_60"],
"explanations": ["Buyer has past-due payments exceeding 60 days"],
"timestamp": "2025-12-30T10:34:00.000Z"
}Request:
curl -X POST http://127.0.0.1:3000/credit-decisions \
-H "Content-Type: application/json" \
-d '{
"buyerId": "BYR-NO-GRADE",
"policyId": "POL-33333",
"requestedLimit": 600000,
"currency": "USD"
}'Response (HTTP 200):
{
"decisionId": "aa0e9400-e29b-41d4-a716-446655440005",
"decision": "REFER",
"approvedLimit": 0,
"currency": "USD",
"reasonCodes": ["RISK_DATA_MISSING"],
"explanations": ["Buyer risk grade data is not available"],
"timestamp": "2025-12-30T10:35:00.000Z"
}Request:
curl -X POST http://127.0.0.1:3000/credit-decisions \
-H "Content-Type: application/json" \
-d '{
"buyerId": "BYR-D-HIGH",
"policyId": "POL-44444",
"requestedLimit": 200000,
"currency": "USD"
}'Response (HTTP 200):
{
"decisionId": "bb0e9400-e29b-41d4-a716-446655440006",
"decision": "DECLINE",
"approvedLimit": 0,
"currency": "USD",
"reasonCodes": ["RISK_GRADE_HIGH"],
"explanations": ["Buyer risk grade is too high (D or E) for credit approval"],
"timestamp": "2025-12-30T10:36:00.000Z"
}Request:
curl -X POST http://127.0.0.1:3000/credit-decisions \
-H "Content-Type: application/json" \
-d '{
"policyId": "POL-55555",
"requestedLimit": 500000,
"currency": "USD"
}'Response (HTTP 400):
{
"errorCode": "MISSING_REQUIRED_FIELD",
"message": "buyerId is required"
}Request:
curl -X POST http://127.0.0.1:3000/credit-decisions \
-H "Content-Type: application/json" \
-d '{
"buyerId": "BYR-12345",
"policyId": "POL-67890",
"requestedLimit": 500000.50,
"currency": "USD"
}'Response (HTTP 400):
{
"errorCode": "INVALID_REQUEST",
"message": "requestedLimit must be a whole number (integer)"
}Request:
curl -X POST http://127.0.0.1:3000/credit-decisions \
-H "Content-Type: application/json" \
-d '{
"buyerId": "BYR-12345",
"policyId": "POL-67890",
"requestedLimit": 500000,
"currency": "USDD"
}'Response (HTTP 400):
{
"errorCode": "INVALID_REQUEST",
"message": "currency must be exactly 3 uppercase letters (e.g., USD, EUR)"
}The MVP uses in-memory stub data. The following buyer IDs are pre-configured:
| Buyer ID | Risk Grade | Past Due >60 | Expected Decision |
|---|---|---|---|
| BYR-A-CLEAN | A | No | APPROVE (up to 1M) |
| BYR-A-PASTDUE | A | Yes | REFER |
| BYR-B-CLEAN | B | No | APPROVE (up to 500K) |
| BYR-B-PASTDUE | B | Yes | REFER |
| BYR-C-MEDIUM | C | No | REFER |
| BYR-D-HIGH | D | No | DECLINE |
| BYR-E-HIGH | E | No | DECLINE |
| BYR-NO-GRADE | None | No | REFER |
| BYR-NO-PASTDUE | A | None | REFER |
| BYR-12345 | A | No | APPROVE (from spec examples) |
| BYR-67890 | B | No | APPROVE (from spec examples) |
Unknown buyers (not in test data) will be treated as having no risk data (REFER decision).
The service implements 7 decision rules evaluated in fixed order:
- Risk grade D or E → DECLINE (approvedLimit = 0)
- Risk grade C → REFER (approvedLimit = 0)
- Risk grade A/B + past due over 60 → REFER (approvedLimit = 0)
- Risk grade A + no past due → APPROVE (up to 1,000,000, cap if needed)
- Risk grade B + no past due → APPROVE (up to 500,000, cap if needed)
- Missing risk grade → REFER (approvedLimit = 0)
- Missing past due data → REFER (approvedLimit = 0)
| Reason Code | Explanation |
|---|---|
| RISK_GRADE_HIGH | Buyer risk grade is too high (D or E) for credit approval |
| RISK_GRADE_MEDIUM | Buyer risk grade (C) requires manual underwriter review |
| PAST_DUE_OVER_60 | Buyer has past-due payments exceeding 60 days |
| LIMIT_CAPPED_BY_GRADE | Requested limit exceeds maximum allowed for buyer's risk grade |
| RISK_DATA_MISSING | Buyer risk grade data is not available |
| PAST_DUE_DATA_MISSING | Buyer past-due payment data is not available |
.
├── src/credit_decision/
│ ├── handler.py # Lambda entrypoint (API Gateway integration)
│ ├── models.py # Dataclasses for request/response/audit
│ ├── decision_engine.py # Pure function decision logic (NO I/O)
│ ├── data_sources.py # Stubbed risk data (in-memory for MVP)
│ ├── audit_repo.py # DynamoDB audit writer
│ └── errors.py # Error types and HTTP mapping
├── tests/
│ ├── test_decision_engine.py # Unit tests for all 7 rules
│ └── test_handler_validation.py # Integration tests for handler
├── template.yaml # AWS SAM Infrastructure as Code
├── requirements.txt # Python dependencies (boto3 only)
├── pyproject.toml # pytest configuration
└── README.md # This file
The MVP uses in-memory stub data in src/credit_decision/data_sources.py:
class InMemoryRiskDataSource:
def get_risk_profile(self, buyer_id: str, policy_id: str) -> BuyerRiskProfile:
# Returns pre-configured test data
return self._test_data.get(buyer_id, ...)To replace stubs with real risk data sources:
-
Implement RiskDataSource Protocol with real integration:
# src/credit_decision/data_sources.py class DynamoDBRiskDataSource: def __init__(self, table_name: str): self.dynamodb = boto3.client('dynamodb') self.table_name = table_name def get_risk_profile(self, buyer_id: str, policy_id: str) -> BuyerRiskProfile: # Real DynamoDB query response = self.dynamodb.get_item( TableName=self.table_name, Key={'pk': {'S': f'BUYER#{buyer_id}'}} ) # Parse and return BuyerRiskProfile ...
-
Update handler.py initialization:
# Replace: risk_data_source = InMemoryRiskDataSource() # With: risk_data_source = DynamoDBRiskDataSource( table_name=os.environ['RISK_DATA_TABLE_NAME'] )
-
Add DynamoDB table and permissions to template.yaml:
Resources: RiskDataTable: Type: AWS::DynamoDB::Table Properties: TableName: BuyerRiskData # ... key schema, billing mode, etc. CreditDecisionFunction: Properties: Environment: Variables: RISK_DATA_TABLE_NAME: !Ref RiskDataTable Policies: - DynamoDBReadPolicy: TableName: !Ref RiskDataTable
No changes needed to decision_engine.py (pure function) or test files (mock the data source).
- External API: Create
ExternalApiRiskDataSourceimplementing the protocol - S3: Create
S3RiskDataSourcefor batch file lookups - RDS: Create
PostgresRiskDataSourcefor relational queries
The RiskDataSource Protocol ensures any implementation works with existing code.
All logs are emitted as JSON with decisionId for correlation:
{
"timestamp": "2025-12-30T10:30:00.000Z",
"level": "INFO",
"decisionId": "550e8400-e29b-41d4-a716-446655440000",
"message": "Decision evaluated",
"decision": "APPROVE",
"approvedLimit": 750000
}Find all decisions for a buyer:
fields @timestamp, decisionId, decision, approvedLimit
| filter buyerId = "BYR-12345"
| sort @timestamp desc
Find all DECLINE decisions:
fields @timestamp, decisionId, buyerId, reasonCodes
| filter decision = "DECLINE"
| sort @timestamp desc
Track error rates:
fields @timestamp, errorCode, message
| filter level = "ERROR"
| stats count() by errorCode
Target: p95 latency <200ms (excluding Lambda cold start)
Actual (measured with SAM local):
- Decision engine (pure function): <5ms
- Full handler execution: 20-50ms (without DynamoDB network latency)
- Cold start: ~1-2 seconds (first invocation only)
Optimization tips:
- Keep Lambda warm with provisioned concurrency for production
- Use DynamoDB on-demand billing for variable load
- Enable X-Ray tracing for detailed performance analysis
Audit Table: Every request (success or failure) writes an immutable record to CreditDecisionAudit DynamoDB table.
Audit Record Schema:
Partition Key: pk = "DECISION#{decisionId}"
Attributes:
- decisionId, timestamp, principalId
- buyerId, policyId, requestedLimit, currency, requestId
- riskGrade, pastDueOver60 (derived inputs)
- decision, approvedLimit, reasonCodes
- status: "OK" | "FAILED"
- errorCode (if failed)
Privacy: No PII stored beyond buyerId and policyId identifiers (per constitution principle II).
pytestpytest --cov=src/credit_decision --cov-report=html
open htmlcov/index.html- decision_engine.py: >90% (pure function, fully tested)
- Overall: >80%
test_decision_engine.py: Unit tests for all 7 decision rules (AC-1 through AC-9, AC-23)test_handler_validation.py: Integration tests for request validation, error handling, audit, logging (AC-10 through AC-22, AC-24, AC-25)
# Ensure Python 3.13 is installed
python3.13 --version
# Clean build artifacts
rm -rf .aws-sam/
sam build# Install dependencies
pip install -r requirements.txt
pip install pytest pytest-cov
# Run from repository root
pytestSAM local uses Docker for DynamoDB. Ensure Docker is running:
docker psFor local testing without DynamoDB, the tests mock the audit repository.
Proprietary - Trade Credit Insurance Application
For questions or issues, contact the credit risk engineering team.