Python SDK

pip install moneyline

from moneyline import Moneyline

client = Moneyline(api_key="ml_live_xxxxx")

# Create a submission
submission = client.submissions.create(
    external_id="acme-001",
    name="Acme Corp",
    metadata={"industry": "retail"},
)

# Upload a document
doc = client.documents.upload(
    submission_id=submission.id,
    file=open("bank-statement.pdf", "rb"),
    filename="bank-statement.pdf",
)

# Start processing
client.submissions.start_workflow(submission.id)

# Wait for completion
submission = client.submissions.wait_for_processing(
    submission.id,
    timeout=300,
    poll_interval=5,
)

# Get analytics
scorecard = client.analytics.scorecard(submission.id)
print(f"Risk score: {scorecard.risk_score}")

from moneyline import AsyncMoneyline

client = AsyncMoneyline(api_key="ml_live_xxxxx")

submission = await client.submissions.create(external_id="acme-001")
doc = await client.documents.upload(
    submission_id=submission.id,
    file=open("statement.pdf", "rb"),
)
scorecard = await client.analytics.scorecard(submission.id)

client = Moneyline(
    api_key="ml_live_xxxxx",          # or set MONEYLINE_API_KEY env var
    base_url="https://api.moneyline.dev",  # or MONEYLINE_BASE_URL
    timeout=60.0,                     # request timeout in seconds
    max_retries=3,                    # automatic retry on 429/5xx
)

# Create
submission = client.submissions.create(
    external_id="acme-001",
    name="Acme Corp",
    metadata={"source": "broker-portal"},
)

# Retrieve
submission = client.submissions.retrieve("sub_xxxxx")

# Update
submission = client.submissions.update("sub_xxxxx", name="Acme Corp Updated")

# List (paginated)
page = client.submissions.list(cursor=None, limit=25)
for sub in page.data:
    print(sub.id, sub.status)

# Delete
client.submissions.delete("sub_xxxxx")

# Start processing workflow
client.submissions.start_workflow("sub_xxxxx")

# Get processing status
status = client.submissions.get_status("sub_xxxxx")

# Enrich submission data
client.submissions.enrich("sub_xxxxx", priority="high")

# Find duplicates
dupes = client.submissions.find_duplicates("sub_xxxxx")

# Sync to CRM
client.submissions.crm_sync("sub_xxxxx")

# Wait for processing to complete
submission = client.submissions.wait_for_processing(
    "sub_xxxxx", timeout=300, poll_interval=5
)

# Upload
doc = client.documents.upload(
    submission_id="sub_xxxxx",
    file=open("statement.pdf", "rb"),
    filename="statement.pdf",
)

# Retrieve
doc = client.documents.retrieve("doc_xxxxx")

# List
docs = client.documents.list(submission_id="sub_xxxxx")

# Delete
client.documents.delete("doc_xxxxx")

# Get raw file content
content = client.documents.get_content("doc_xxxxx")

# Get parsed/extracted data
parsed = client.documents.get_parsed_results("doc_xxxxx")

# Reclassify document type
doc = client.documents.reclassify("doc_xxxxx", document_class="bank_statement")

# Re-parse document
doc = client.documents.parse("doc_xxxxx")

# Split multi-page document
split_docs = client.documents.split("doc_xxxxx")

# Export all docs as XLSX
xlsx_bytes = client.documents.export("sub_xxxxx", format="xlsx")

# Enrich transactions
enriched = client.transactions.enrich(
    transactions=[
        {"description": "AMZN MKTP US*123", "amount": -49.99, "date": "2024-01-15"},
        {"description": "DIRECT DEP ACME CO", "amount": 5000.00, "date": "2024-01-14"},
    ]
)

# Create
txn = client.transactions.create(
    submission_id="sub_xxxxx",
    description="Wire Transfer",
    amount=10000.00,
    date="2024-03-01",
)

# List
txns = client.transactions.list(submission_id="sub_xxxxx", limit=100)

# Delete
client.transactions.delete("txn_xxxxx")

# Stream enrichment (large batches)
stream = client.transactions.stream_enrichment(transactions=[...])
for chunk in stream:
    print(chunk)

# Full scorecard
scorecard = client.analytics.scorecard("sub_xxxxx")

# Request scorecard generation (async processing)
client.analytics.request_scorecard("sub_xxxxx")

# Cashflow analysis
cashflow = client.analytics.cashflow("sub_xxxxx")

# Risk score
risk = client.analytics.risk_score("sub_xxxxx")

# Account balances
balances = client.analytics.balances("sub_xxxxx")

# Anomaly detection
anomalies = client.analytics.anomalies("sub_xxxxx")

# Forecasts
forecasts = client.analytics.forecasts("sub_xxxxx")

# Statistical summary
stats = client.analytics.statistics("sub_xxxxx")

# Bank statement summary
summary = client.analytics.bank_summary("sub_xxxxx")

# Recurring transactions
recurring = client.analytics.recurring("sub_xxxxx")

# Merchant summary
merchants = client.analytics.merchant_summary("sub_xxxxx")

# Benchmarks
benchmarks = client.analytics.benchmarks("sub_xxxxx")

# Data coverage report
coverage = client.analytics.coverage("sub_xxxxx")

# Grouped transactions
grouped = client.analytics.grouped_transactions("sub_xxxxx", group_by="merchant")

# Full financials
financials = client.analytics.financials("sub_xxxxx")

# Portfolio-wide decline analytics
decline = client.analytics.decline_analytics()

# Export analytics as bytes
export = client.analytics.export("sub_xxxxx")

# Create
webhook = client.webhooks.create(
    url="https://example.com/webhooks/moneyline",
    events=["submission.completed", "document.parsed"],
)

# Retrieve
webhook = client.webhooks.retrieve("wh_xxxxx")

# List
webhooks = client.webhooks.list(limit=25)

# Update
webhook = client.webhooks.update("wh_xxxxx", url="https://new-url.com/hook")

# Delete
client.webhooks.delete("wh_xxxxx")

# Send test event
client.webhooks.test("wh_xxxxx", event_type="submission.completed")

# List deliveries
deliveries = client.webhooks.deliveries("wh_xxxxx")

# Get available event types
events = client.webhooks.event_types()

# Policies
policy = client.policies.create(name="Auto-approve <$50k", rules=[...])
policies = client.policies.list()
client.policies.evaluate("sub_xxxxx")

# Rules
rule = client.rules.create(
    policy_id="pol_xxxxx",
    field="risk_score",
    operator="lt",
    value=30,
    action="approve",
)
rules = client.rules.list(policy_id="pol_xxxxx")

merchants = client.merchants.search("Amazon")
merchant = client.merchants.retrieve("mer_xxxxx")

categories = client.categories.list()

funder = client.funders.retrieve("fnd_xxxxx")
linked = client.funders.list_linked()
result = client.funders.submit(submission_id="sub_xxxxx", funder_id="fnd_xxxxx")

integration = client.integrations.create(type="plaid", name="Plaid Prod")
link = client.integrations.create_link("int_xxxxx")
links = client.integrations.list_links("int_xxxxx")
client.integrations.refresh_link("link_xxxxx")
client.integrations.delete_link("link_xxxxx")

definitions = client.entities.list_definitions()
entities = client.entities.list_for_submission("sub_xxxxx")
client.entities.bulk_save_attributes([...])

iso = client.iso_applications.upload(
    file=open("iso-app.pdf", "rb"),
    filename="iso-app.pdf",
    submission_id="sub_xxxxx",
)

from moneyline import Webhook

payload = request.body  # raw bytes
signature = request.headers["X-Moneyline-Signature"]

event = Webhook.verify(
    payload=payload,
    signature=signature,
    secret="whsec_xxxxx",
)
print(event.type)  # "submission.completed"

from moneyline import MoneylineError, RateLimitError

try:
    submission = client.submissions.retrieve("sub_xxxxx")
except RateLimitError as e:
    print(f"Rate limited, retry after {e.retry_after}s")
except MoneylineError as e:
    print(f"API error: {e.code} — {e.message}")

# Auto-paginate all results
for submission in client.submissions.list().auto_paging_iter():
    print(submission.id)