How to Build a Real Estate Transaction Management System

A real estate transaction management automation system tracks every deal from accepted offer to closing day. I built this for a team that was losing track of deadlines buried in email threads. Now every milestone has a due date, every document has a status, and nothing falls through the cracks.

What You Need Before Starting

• Python 3.8+
• SQLite
• Email sending capability for notifications
• A checklist of your standard transaction milestones

Step 1: Define Transaction Milestones

TRANSACTION_MILESTONES = [
    {"name": "offer_accepted", "offset_days": 0, "required_docs": ["purchase_agreement"]},
    {"name": "earnest_money_deposited", "offset_days": 3, "required_docs": ["deposit_receipt"]},
    {"name": "inspection_scheduled", "offset_days": 5, "required_docs": []},
    {"name": "inspection_complete", "offset_days": 10, "required_docs": ["inspection_report"]},
    {"name": "inspection_response", "offset_days": 12, "required_docs": ["repair_request"]},
    {"name": "appraisal_ordered", "offset_days": 7, "required_docs": []},
    {"name": "appraisal_complete", "offset_days": 21, "required_docs": ["appraisal_report"]},
    {"name": "loan_approval", "offset_days": 25, "required_docs": ["commitment_letter"]},
    {"name": "title_clear", "offset_days": 28, "required_docs": ["title_commitment"]},
    {"name": "final_walkthrough", "offset_days": 29, "required_docs": []},
    {"name": "closing", "offset_days": 30, "required_docs": ["closing_disclosure", "deed"]},
]

Step 2: Set Up the Transaction Database

import sqlite3
from datetime import datetime, timedelta

def init_transaction_db(db_path="transactions.db"):
    conn = sqlite3.connect(db_path)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS transactions (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            property_address TEXT,
            buyer_name TEXT,
            seller_name TEXT,
            sale_price REAL,
            offer_accepted_date TEXT,
            closing_date TEXT,
            status TEXT DEFAULT 'active',
            agent TEXT,
            created_at TEXT
        )
    """)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS milestones (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            transaction_id INTEGER,
            milestone_name TEXT,
            due_date TEXT,
            completed_date TEXT,
            status TEXT DEFAULT 'pending',
            notes TEXT
        )
    """)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS documents (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            transaction_id INTEGER,
            milestone_name TEXT,
            doc_name TEXT,
            file_path TEXT,
            uploaded_at TEXT,
            status TEXT DEFAULT 'missing'
        )
    """)
    conn.commit()
    conn.close()

def create_transaction(property_address, buyer, seller, price, offer_date, agent, db_path="transactions.db"):
    conn = sqlite3.connect(db_path)
    offer_dt = datetime.fromisoformat(offer_date)
    closing_date = (offer_dt + timedelta(days=30)).isoformat()[:10]
    
    conn.execute("""
        INSERT INTO transactions (property_address, buyer_name, seller_name, sale_price,
            offer_accepted_date, closing_date, agent, created_at)
        VALUES (?,?,?,?,?,?,?,?)
    """, (property_address, buyer, seller, price, offer_date, closing_date, agent, datetime.utcnow().isoformat()))
    
    txn_id = conn.execute("SELECT last_insert_rowid()").fetchone()[0]
    
    for milestone in TRANSACTION_MILESTONES:
        due = (offer_dt + timedelta(days=milestone["offset_days"])).isoformat()[:10]
        conn.execute(
            "INSERT INTO milestones (transaction_id, milestone_name, due_date) VALUES (?,?,?)",
            (txn_id, milestone["name"], due)
        )
        for doc in milestone["required_docs"]:
            conn.execute(
                "INSERT INTO documents (transaction_id, milestone_name, doc_name) VALUES (?,?,?)",
                (txn_id, milestone["name"], doc)
            )
    
    conn.commit()
    conn.close()
    return txn_id

Step 3: Build Deadline Monitoring

def check_upcoming_deadlines(days_ahead=3, db_path="transactions.db"):
    conn = sqlite3.connect(db_path)
    upcoming = conn.execute("""
        SELECT m.milestone_name, m.due_date, t.property_address, t.agent
        FROM milestones m
        JOIN transactions t ON m.transaction_id = t.id
        WHERE m.status = 'pending'
        AND m.due_date BETWEEN date('now') AND date('now', ?)
        AND t.status = 'active'
        ORDER BY m.due_date
    """, (f"+{days_ahead} days",)).fetchall()
    conn.close()
    
    return [{"milestone": u[0], "due": u[1], "property": u[2], "agent": u[3]} for u in upcoming]

def check_overdue_milestones(db_path="transactions.db"):
    conn = sqlite3.connect(db_path)
    overdue = conn.execute("""
        SELECT m.milestone_name, m.due_date, t.property_address, t.agent
        FROM milestones m
        JOIN transactions t ON m.transaction_id = t.id
        WHERE m.status = 'pending' AND m.due_date < date('now') AND t.status = 'active'
    """).fetchall()
    conn.close()
    return overdue

Step 4: Missing Document Alerts

def check_missing_documents(db_path="transactions.db"):
    conn = sqlite3.connect(db_path)
    missing = conn.execute("""
        SELECT d.doc_name, d.milestone_name, t.property_address, m.due_date
        FROM documents d
        JOIN milestones m ON d.transaction_id = m.transaction_id AND d.milestone_name = m.milestone_name
        JOIN transactions t ON d.transaction_id = t.id
        WHERE d.status = 'missing'
        AND m.due_date <= date('now', '+3 days')
        AND t.status = 'active'
    """).fetchall()
    conn.close()
    return missing

Step 5: Daily Transaction Report

import requests

def daily_transaction_report(db_path="transactions.db"):
    upcoming = check_upcoming_deadlines(3, db_path)
    overdue = check_overdue_milestones(db_path)
    missing_docs = check_missing_documents(db_path)
    
    lines = ["Transaction Management Daily Report", ""]
    
    if overdue:
        lines.append(f"OVERDUE ({len(overdue)}):")
        for o in overdue:
            lines.append(f"  {o[2]}: {o[0]} was due {o[1]}")
    
    if upcoming:
        lines.append(f"\nUpcoming ({len(upcoming)}):")
        for u in upcoming:
            lines.append(f"  {u['property']}: {u['milestone']} due {u['due']}")
    
    if missing_docs:
        lines.append(f"\nMissing Documents ({len(missing_docs)}):")
        for d in missing_docs:
            lines.append(f"  {d[2]}: {d[0]} needed for {d[1]} (due {d[3]})")
    
    report = "\n".join(lines)
    requests.post("YOUR_SLACK_WEBHOOK", json={"text": report})
    return report

if __name__ == "__main__":
    print(daily_transaction_report())

What to Build Next

Add a client-facing status page where buyers and sellers can see their transaction progress without calling the agent for updates.

Related Reading

• AI for Workflow Optimization - automating multi-step workflows
• AI in Real Estate Operations - AI across real estate
• Why Process Documentation Is the First Step - documenting transaction processes