Why Financial Data Pipelines Are a Product Goldmine
Data analysts have a shared pain point: every new project starts from zero code.
Client A needs financial reports — you write CSV reading, aggregation, and report generation code. Client B also needs it — you copy and tweak paths. Client C wants invoice data added — you add another table. Three months later, you have a dozen “similar but different” projects, each requiring separate maintenance.
The root cause isn’t bad code — it’s missing reusable architecture.
DuckDB’s VIEW consolidation pattern + Python class encapsulation lets you write an engine once and reuse it across N clients. This article breaks down the architecture and how to turn it into a ¥299/month × N revenue product.
1. Architecture Overview: Three-Layer Data Pipeline
A reusable financial data pipeline has three layers:
┌─────────────────────────────────────────────┐
│ Layer 1: Data Ingestion │
│ bank_transactions.csv → read_csv │
│ invoices.csv → read_csv │
│ expenses.csv → read_csv │
└──────────────────┬──────────────────────────┘
▼
┌─────────────────────────────────────────────┐
│ Layer 2: View Consolidation │
│ v_all_income = UNION ALL (bank + invoices)│
│ v_all_expense = UNION ALL (bank + platforms)│
└──────────────────┬──────────────────────────┘
▼
┌─────────────────────────────────────────────┐
│ Layer 3: Analysis Engine │
│ P&L / Cash Flow / Tax / Trend Analysis │
└─────────────────────────────────────────────┘
Key design principle: Layer 1 and Layer 2 are identical for every client. Only Layer 3 business rules need customization.
This means you only adjust analysis logic per client — no rewriting data ingestion and consolidation code.
2. Data Ingestion Layer: Multi-Source CSV in One SQL
Small businesses have data scattered everywhere: bank export CSVs, invoice system CSVs, Alipay/WeChat merchant platform CSVs. Traditional approach requires writing three different parsing scripts. With DuckDB, one SQL handles everything.
import duckdb
from pathlib import Path
class FinancialPipeline:
"""Reusable financial data pipeline engine"""
def __init__(self, db_path: str):
self.con = duckdb.connect(db_path)
self._create_schema()
def _create_schema(self):
"""Standard schema: shared across all projects"""
self.con.execute("""
-- Bank transactions: exported from online banking
CREATE TABLE IF NOT EXISTS bank_transactions (
date DATE,
type VARCHAR,
category VARCHAR,
amount DECIMAL(12,2),
counterparty VARCHAR,
source VARCHAR
)
""")
self.con.execute("""
-- Invoice data: from invoice management system
CREATE TABLE IF NOT EXISTS invoices (
invoice_date DATE,
invoice_number VARCHAR,
seller VARCHAR,
buyer VARCHAR,
amount DECIMAL(12,2),
tax_rate DECIMAL(5,4),
status VARCHAR
)
""")
self.con.execute("""
-- Platform expenses: Alipay / WeChat / bank transfer
CREATE TABLE IF NOT EXISTS platform_expenses (
expense_date DATE,
platform VARCHAR,
category VARCHAR,
amount DECIMAL(12,2),
vendor VARCHAR,
remark VARCHAR
)
""")
Architecture highlight: Schema defined once, reused across all projects. For new clients, just COPY ... FROM 'data/*.csv' to import data — no table structure changes needed.
3. View Consolidation Layer: UNION ALL for Unified Income/Expense口径
This is the core of the entire architecture. Different data sources have different semantics for “income” and “expense”:
- Bank transactions with
type='income'are income - Invoices with
status='已收款'(collected) are also income - Alipay expenses and bank expenses are the same thing
Use VIEWs to unify them into the same口径:
-- Unified income view: bank income + collected invoices
CREATE OR REPLACE VIEW v_all_income AS
SELECT
date AS trans_date,
'bank' AS source_type,
category,
amount,
counterparty AS partner
FROM bank_transactions
WHERE type = 'income'
UNION ALL
SELECT
invoice_date AS trans_date,
'invoice_system' AS source_type,
'sales' AS category,
amount,
buyer AS partner
FROM invoices
WHERE status = 'collected';
-- Unified expense view: bank expenses + platform expenses
CREATE OR REPLACE VIEW v_all_expense AS
SELECT
date AS trans_date,
'bank' AS source_type,
category,
amount,
counterparty AS partner
FROM bank_transactions
WHERE type = 'expense'
UNION ALL
SELECT
expense_date AS trans_date,
'platform' AS source_type,
category,
amount,
vendor AS partner
FROM platform_expenses;
Why VIEWs instead of physical tables?
- Real-time freshness: VIEWs read from base tables on every query — no re-ingest needed when new data arrives
- Storage efficient: No duplication, income and expense stored only once
- Centralized logic: Consolidation rules written once, all analysis queries benefit automatically
- Traceable: Each field has
source_type, enabling data lineage tracking
4. Analysis Engine Layer: One-Click P&L Generation
With unified income/expense views, analysis logic becomes straightforward:
-- Monthly P&L statement
CREATE OR REPLACE VIEW v_monthly_pl AS
WITH monthly_income AS (
SELECT
DATE_TRUNC('month', trans_date) AS month,
SUM(amount) AS total_income,
COUNT(*) AS transaction_count
FROM v_all_income
GROUP BY 1
),
monthly_expense AS (
SELECT
DATE_TRUNC('month', trans_date) AS month,
SUM(amount) AS total_expense,
COUNT(*) AS transaction_count
FROM v_all_expense
GROUP BY 1
),
monthly_tax AS (
-- Extract estimated taxes from invoices
SELECT
DATE_TRUNC('month', invoice_date) AS month,
SUM(amount * tax_rate) AS estimated_tax
FROM invoices
WHERE status IN ('issued', 'collected')
GROUP BY 1
)
SELECT
i.month,
ROUND(i.total_income, 2) AS revenue,
ROUND(e.total_expense, 2) AS expenses,
ROUND(i.total_income - e.total_expense, 2) AS gross_profit,
ROUND((i.total_income - e.total_expense) / NULLIF(i.total_income, 0) * 100, 2) AS profit_margin_pct,
ROUND(COALESCE(t.estimated_tax, 0), 2) AS estimated_tax,
ROUND(i.total_income - e.total_expense - COALESCE(t.estimated_tax, 0), 2) AS net_profit,
i.transaction_count + e.transaction_count AS total_transactions
FROM monthly_income i
LEFT JOIN monthly_expense e ON i.month = e.month
LEFT JOIN monthly_tax t ON i.month = t.month;
Add MoM and YoY growth (one SQL, no pandas merge needed):
-- Full P&L with MoM and YoY
SELECT
month,
revenue,
expenses,
gross_profit,
profit_margin_pct,
-- Previous month (MoM)
LAG(revenue) OVER w AS prev_month_revenue,
ROUND(
(revenue - LAG(revenue) OVER w)
/ NULLIF(LAG(revenue) OVER w, 0) * 100, 2
) AS mom_growth_pct,
-- Same month last year (YoY)
LAG(revenue, 12) OVER w AS same_month_last_year,
ROUND(
(revenue - LAG(revenue, 12) OVER w)
/ NULLIF(LAG(revenue, 12) OVER w, 0) * 100, 2
) AS yoy_growth_pct
FROM v_monthly_pl
WINDOW w AS (ORDER BY month);
5. Python Encapsulation: One Engine, N Clients
Complete reusable engine class:
class FinancialPipeline:
"""Financial data pipeline engine — write once, use N times"""
def __init__(self, db_path: str):
self.con = duckdb.connect(db_path)
self._setup_views()
def _setup_views(self):
"""View consolidation layer: shared across all projects"""
# v_all_income and v_all_expense from above
pass
def ingest(self, month: str, data_dir: Path):
"""Data ingestion: called once per month"""
for csv_file in data_dir.glob(f"{month}*.csv"):
table_name = csv_file.stem
self.con.execute(f"""
COPY {table_name} FROM '{csv_file}'
(FORMAT CSV, HEADER, AUTO_DETERMINE)
""")
print(f"✅ {month} data ingested")
def generate_report(self, month: str) -> pd.DataFrame:
"""Generate monthly P&L statement"""
return self.con.execute(f"""
SELECT * FROM v_monthly_pl
WHERE month = '{month}-01'::DATE
""").fetchdf()
def export_parquet(self, month: str, output_dir: Path):
"""Export as Parquet — 5x faster subsequent queries"""
df = self.generate_report(month)
df.to_parquet(output_dir / f"pl_{month}.parquet")
def close(self):
self.con.close()
Usage:
# Client A: Restaurant
pipeline_a = FinancialPipeline("clients/restaurant_01.duckdb")
pipeline_a.ingest("2026-07", Path("clients/restaurant_01/data"))
report_a = pipeline_a.generate_report("2026-07")
pipeline_a.export_parquet("2026-07", Path("clients/restaurant_01/output"))
pipeline_a.close()
# Client B: E-commerce — same engine, different data
pipeline_b = FinancialPipeline("clients/ecommerce_02.duckdb")
pipeline_b.ingest("2026-07", Path("clients/ecommerce_02/data"))
report_b = pipeline_b.generate_report("2026-07")
pipeline_b.close()
Key advantage: Client A and Client B use the same code. Each client has an independent DuckDB database file. Adding a new client = creating a new folder + importing data, zero code changes.
6. DuckDB vs Traditional Approach Comparison
| Dimension | DuckDB Pipeline | pandas + Manual | Traditional ETL |
|---|---|---|---|
| Multi-source CSV loading | One SQL COPY | Multiple pd.read_csv + merge | Complex configuration |
| VIEW consolidation | Native SQL VIEW | Multiple DataFrame merges | Depends on scheduler |
| MoM/YoY growth | LAG() window function | Need self-merge | Custom scripting |
| Engine reusability | Class-based, zero-code reuse | Copy-paste | Re-configuration needed |
| Deployment | Zero (single .duckdb file) | Requires Python env | Requires server |
7. Monetization Paths: From 0 to ¥10,000+/month
Path 1: Monthly Subscription (Recommended)
- ¥299/client/month for automated monthly report generation
- 5 clients = ¥1,495/month
- 10 clients = ¥2,990/month
- Marginal cost ≈ 0 (DuckDB processes 100K rows in < 1 second)
Path 2: Per-Project Fee
- One-time fee ¥3,000-8,000 per client
- Includes initial data migration + custom analysis logic
- Ongoing maintenance billed separately at ¥500/month
Path 3: Productized SaaS
- Wrap with FastAPI as a web application
- Clients self-upload data, view reports in real-time
- Price: ¥99-299/month/client
- Multi-tenant support (each client gets independent DuckDB file)
Key insight: Financial reports are a must-have. Small businesses either hire an accountant (monthly salary ¥6,000-12,000) or buy accounting software (annual fee ¥2,000-5,000). Your DuckDB solution at ¥299/month is 1/10 the price of accounting software with equivalent results.
8. Production-Grade: Data Quality Checks
A real production pipeline needs data quality validation:
-- Data quality check view
CREATE OR REPLACE VIEW v_data_quality AS
SELECT
'bank_transactions' AS table_name,
COUNT(*) AS total_rows,
SUM(CASE WHEN amount IS NULL THEN 1 ELSE 0 END) AS null_amounts,
SUM(CASE WHEN amount < 0 THEN 1 ELSE 0 END) AS negative_amounts,
SUM(amount) AS total_amount
FROM bank_transactions
UNION ALL
SELECT
'invoices' AS table_name,
COUNT(*) AS total_rows,
SUM(CASE WHEN amount IS NULL THEN 1 ELSE 0 END),
SUM(CASE WHEN amount < 0 THEN 1 ELSE 0 END),
SUM(amount)
FROM invoices;
With quality checks, you can automatically validate data integrity before delivery. This is your charging justification — “I don’t just give you reports, I guarantee data quality”.
Summary
The core value of this architecture isn’t the technology itself — it’s reusability:
- VIEW consolidation pattern: Unifies multi-source data semantics, write rules once, all analyses benefit automatically
- Class-based engine: One
FinancialPipelineclass serves N clients, zero code for new clients - DuckDB single file: Each client gets one .duckdb file — distribution, backup, and migration are trivial
- Subscription monetization: ¥299/month, marginal cost near zero, true passive income
After this article, you have all the skills needed to build financial data products. The next step is finding a real client and running the full pipeline with real data.
💡 More DuckDB pipeline patterns → duckdblab.org