Automated Investment Research Briefing with DuckDB: From Multi-Source Data to Daily Reports

import duckdb
import pandas as pd
from pathlib import Path

# One in-memory database, directly reading all CSVs
conn = duckdb.connect(':memory:')

# read_csv_auto auto-infers dates, numeric types — no manual format handling needed
conn.execute("CREATE TABLE stocks AS SELECT * FROM read_csv_auto('data/stock_prices.csv')")
conn.execute("CREATE TABLE financials AS SELECT * FROM read_csv_auto('data/financials.csv')")
conn.execute("CREATE TABLE macro AS SELECT * FROM read_csv_auto('data/macro.csv')")

print(f"✅ stocks: {conn.execute('SELECT COUNT(*) FROM stocks').fetchone()[0]} rows")
print(f"✅ financials: {conn.execute('SELECT COUNT(*) FROM financials').fetchone()[0]} rows")
print(f"✅ macro: {conn.execute('SELECT COUNT(*) FROM macro').fetchone()[0]} rows")

WITH pe_rank AS (
    SELECT 
        f.symbol,
        f.name,
        f.pe_ttm,
        f.roe,
        f.revenue_yoy,
        f.sector,
        -- Sector PE average (window function, no GROUP BY needed)
        AVG(f.pe_ttm) OVER (PARTITION BY f.sector) AS sector_pe_avg,
        -- PE rank within sector
        RANK() OVER (PARTITION BY f.sector ORDER BY f.pe_ttm) AS pe_rank_in_sector
    FROM financials f
    WHERE f.pe_ttm > 0 AND f.roe > 0
)

momentum AS (
    SELECT 
        symbol,
        ((close - LAG(close, 20) OVER (PARTITION BY symbol ORDER BY date)) 
         / LAG(close, 20) OVER (PARTITION BY symbol ORDER BY date)) * 100 AS momentum_20d
    FROM stocks
    WHERE date = (SELECT MAX(date) FROM stocks)
)

macro_filter AS (
    SELECT 
        date,
        cpi,
        pmi,
        rate,
        CASE 
            WHEN pmi > 50 AND rate <= LAG(rate) OVER (ORDER BY date) THEN 'risk_on'
            WHEN pmi < 49 AND rate >= LAG(rate) OVER (ORDER BY date) THEN 'risk_off'
            ELSE 'neutral'
        END AS market_regime
    FROM macro
    ORDER BY date DESC
    LIMIT 1
)

SELECT 
    p.symbol,
    p.name,
    p.sector,
    ROUND(p.pe_ttm, 2) AS pe,
    ROUND(p.sector_pe_avg, 2) AS sector_pe_avg,
    ROUND(p.roe, 2) AS roe_pct,
    ROUND(m.momentum_20d, 2) AS momentum_20d,
    ROUND(r.cpi, 3) AS cpi,
    r.pmi,
    r.market_regime
FROM pe_rank p
LEFT JOIN momentum m ON p.symbol = m.symbol
CROSS JOIN macro_filter r
WHERE p.pe_ttm < p.sector_pe_avg
  AND p.roe >= 15
  AND p.pe_rank_in_sector <= 5
ORDER BY p.pe_ttm ASC
LIMIT 20

from jinja2 import Template
from datetime import datetime

template_text = """
📊 Investment Research Briefing | {{ date }}

━━━━━━━━━━━━━━━━━━
【Macro Environment】
CPI: {{ cpi }} | PMI: {{ pmi }} | Regime: {{ regime }}

━━━━━━━━━━━━━━━━━━
【Low Valuation, High ROE Picks】(PE < Sector Avg & ROE ≥ 15%)

{% for row in stocks %}
{{ loop.index }}. {{ row.name }}（{{ row.symbol }}）
   Sector: {{ row.sector }} | PE: {{ row.pe }} (Sector Avg: {{ row.sector_pe_avg }})
   ROE: {{ row.roe_pct }}% | 20-Day Momentum: {{ row.momentum_20d }}%
{% endfor %}

━━━━━━━━━━━━━━━━━━
⚠️ Data as of {{ date }}, for reference only, not investment advice.
Generated by DuckDB Auto-Report
"""

template = Template(template_text)
report = template.render(
    date=datetime.now().strftime('%Y-%m-%d'),
    cpi=float(screener['cpi'].iloc[0]) if len(screener) > 0 else None,
    pmi=float(screener['pmi'].iloc[0]) if len(screener) > 0 else None,
    regime=str(screener['market_regime'].iloc[0]) if len(screener) > 0 else 'unknown',
    stocks=[
        {
            'name': r['name'],
            'symbol': r['symbol'],
            'sector': r['sector'],
            'pe': r['pe'],
            'sector_pe_avg': r['sector_pe_avg'],
            'roe_pct': r['roe_pct'],
            'momentum_20d': r['momentum_20d'] if pd.notna(r['momentum_20d']) else '--',
        }
        for _, r in screener.iterrows()
    ]
)

print(report)

# auto_report.py — complete entry point
import duckdb
from jinja2 import Template
from datetime import datetime
import smtplib
from email.mime.text import MIMEText
import os

EMAIL_CONFIG = {
    'smtp_host': os.environ.get('SMTP_HOST'),
    'smtp_port': int(os.environ.get('SMTP_PORT', '587')),
    'sender': os.environ.get('EMAIL_USER'),
    'password': os.environ.get('EMAIL_PASS'),
    'recipients': os.environ.get('EMAIL_RECIPIENTS', '').split(','),
}

def generate_report() -> str:
    conn = duckdb.connect(':memory:')
    
    # Read data
    conn.execute("CREATE TABLE stocks AS SELECT * FROM read_csv_auto('data/stock_prices.csv')")
    conn.execute("CREATE TABLE financials AS SELECT * FROM read_csv_auto('data/financials.csv')")
    conn.execute("CREATE TABLE macro AS SELECT * FROM read_csv_auto('data/macro.csv')")
    
    # Execute analysis query
    query = """
    WITH pe_rank AS (
        SELECT symbol, name, pe_ttm, roe, revenue_yoy, sector,
               AVG(pe_ttm) OVER (PARTITION BY sector) AS sector_pe_avg,
               RANK() OVER (PARTITION BY sector ORDER BY pe_ttm) AS pe_rank_in_sector
        FROM financials WHERE pe_ttm > 0 AND roe > 0
    ),
    momentum AS (
        SELECT symbol,
               ((close - LAG(close, 20) OVER (PARTITION BY symbol ORDER BY date)) 
                / LAG(close, 20) OVER (PARTITION BY symbol ORDER BY date)) * 100 AS momentum_20d
        FROM stocks
        WHERE date = (SELECT MAX(date) FROM stocks)
    ),
    macro_filter AS (
        SELECT date, cpi, pmi, rate,
               CASE 
                   WHEN pmi > 50 AND rate <= LAG(rate) OVER (ORDER BY date) THEN 'risk_on'
                   WHEN pmi < 49 AND rate >= LAG(rate) OVER (ORDER BY date) THEN 'risk_off'
                   ELSE 'neutral'
               END AS market_regime
        FROM macro ORDER BY date DESC LIMIT 1
    )
    SELECT p.symbol, p.name, p.sector, ROUND(p.pe_ttm,2) AS pe,
           ROUND(p.sector_pe_avg,2) AS sector_pe_avg, ROUND(p.roe,2) AS roe_pct,
           ROUND(m.momentum_20d,2) AS momentum_20d,
           ROUND(r.cpi,3) AS cpi, r.pmi, r.market_regime
    FROM pe_rank p
    LEFT JOIN momentum m ON p.symbol = m.symbol
    CROSS JOIN macro_filter r
    WHERE p.pe_ttm < p.sector_pe_avg AND p.roe >= 15 AND p.pe_rank_in_sector <= 5
    ORDER BY p.pe_ttm ASC LIMIT 20
    """
    
    screener = conn.execute(query).fetchdf()
    conn.close()
    
    # Render report
    template = Template("""
📊 Investment Research Briefing | {{ date }}
━━━━━━━━━━━━━━━━━━
【Macro】CPI: {{ cpi }} | PMI: {{ pmi }} | Regime: {{ regime }}
━━━━━━━━━━━━━━━━━━
【Low Valuation, High ROE Picks】(PE < Sector Avg & ROE ≥ 15%)
{% for row in stocks %}
{{ loop.index }}. {{ row.name }}（{{ row.symbol }}）
   Sector: {{ row.sector }} | PE: {{ row.pe }} (Avg: {{ row.sector_pe_avg }})
   ROE: {{ row.roe_pct }}% | 20d Momentum: {{ row.momentum_20d }}%
{% endfor %}
━━━━━━━━━━━━━━━━━━
⚠️ Data as of {{ date }}, for reference only.
Generated by DuckDB Auto-Report
    """)
    
    return template.render(
        date=datetime.now().strftime('%Y-%m-%d'),
        cpi=float(screener['cpi'].iloc[0]) if len(screener) > 0 else None,
        pmi=float(screener['pmi'].iloc[0]) if len(screener) > 0 else None,
        regime=str(screener['market_regime'].iloc[0]) if len(screener) > 0 else 'unknown',
        stocks=[
            {'name': r['name'], 'symbol': r['symbol'], 'sector': r['sector'],
             'pe': r['pe'], 'sector_pe_avg': r['sector_pe_avg'],
             'roe_pct': r['roe_pct'],
             'momentum_20d': r['momentum_20d'] if pd.notna(r['momentum_20d']) else '--'}
            for _, r in screener.iterrows()
        ]
    )

def send_report(report: str):
    msg = MIMEText(report, 'plain', 'utf-8')
    msg['Subject'] = f'Investment Research Briefing | {datetime.now().strftime("%Y-%m-%d")}'
    msg['From'] = EMAIL_CONFIG['sender']
    msg['To'] = ', '.join(EMAIL_CONFIG['recipients'])
    with smtplib.SMTP(EMAIL_CONFIG['smtp_host'], EMAIL_CONFIG['smtp_port']) as s:
        s.starttls()
        s.login(EMAIL_CONFIG['sender'], EMAIL_CONFIG['password'])
        s.sendmail(EMAIL_CONFIG['sender'], EMAIL_CONFIG['recipients'], msg.as_string())

if __name__ == '__main__':
    report = generate_report()
    send_report(report)
    print('✅ Report sent')