Building an Automated Stock Analysis Report with DuckDB: From Data Fetch to Report Generation

pip install yfinance duckdb pandas matplotlib

import yfinance as yf
import duckdb
from datetime import datetime, timedelta
import os

TICKERS = ["AAPL", "MSFT", "GOOGL", "AMZN", "NVDA", "TSLA", "META"]
DATA_DIR = "stock_data"
os.makedirs(DATA_DIR, exist_ok=True)

def fetch_and_save():
 end_date = datetime.now()
 start_date = end_date - timedelta(days=400)

 for ticker in TICKERS:
 df = yf.download(ticker, start=start_date, end=end_date, progress=False)
 if df.empty:
 continue
 csv_path = os.path.join(DATA_DIR, f"{ticker}.csv")
 df.to_csv(csv_path)
 print(f"✅ {ticker} data saved: {csv_path}")

 # Load CSVs into DuckDB
 conn = duckdb.connect("stocks.db")
 conn.execute("DROP TABLE IF EXISTS metadata")
 tables = []
 for ticker in TICKERS:
 csv_path = os.path.join(DATA_DIR, f"{ticker}.csv")
 if os.path.exists(csv_path):
 conn.execute(f"CREATE TABLE {ticker} AS SELECT * FROM read_csv_auto('{csv_path}')")
 tables.append(ticker)
 conn.execute("CREATE TABLE metadata AS SELECT 'stocks' as source, current_date() as last_updated")
 conn.commit()
 conn.close()
 print(f"🗄️ Database updated, {len(tables)} stocks loaded")

if __name__ == "__main__":
 fetch_and_save()

import duckdb

conn = duckdb.connect("stocks.db")

top_gainers = conn.execute("""
 SELECT ticker,
 first(close) as close_30d_ago,
 last(close) as close_current,
 round((last(close) - first(close)) / first(close) * 100, 2) as pct_change
 FROM (
 SELECT ticker, date, close,
 ROW_NUMBER() OVER (PARTITION BY ticker ORDER BY date DESC) as rn_desc,
 ROW_NUMBER() OVER (PARTITION BY ticker ORDER BY date ASC) as rn_asc
 FROM (
 UNPIVOT stocks ON columns INCLUDE (date) INTO name, value
 )
 WHERE name = 'close'
 ) sub
 WHERE rn_desc <= 1 OR rn_asc <= 1
 GROUP BY ticker
 ORDER BY pct_change DESC
 LIMIT 5
""").fetchdf()

print("📈 Top 5 Gainers (30-day):")
for _, row in top_gainers.iterrows():
 print(f" {row.ticker}: {row.pct_change}%")

volatility = conn.execute("""
 SELECT ticker,
 round(stddev(close), 2) as volatility,
 round(avg(volume) / 1e6, 1) as avg_volume_m,
 round(avg(close), 2) as avg_price
 FROM (
 SELECT ticker, date, close, volume
 FROM (
 UNPIVOT stocks ON columns INCLUDE (date) INTO name, value
 )
 WHERE name IN ('close', 'volume')
 PIVOT_agg(avg) ON name
 )
 GROUP BY ticker
 ORDER BY volatility DESC
 LIMIT 5
""").fetchdf()

print("\n📊 Top 5 by Volatility (High Risk, High Reward):")
for _, row in volatility.iterrows():
 print(f" {row.ticker}: Volatility={row.volatility}, Avg Price=${row.avg_price}")

ma_signals = conn.execute("""
 WITH daily AS (
 SELECT ticker, date, close,
 AVG(close) OVER (PARTITION BY ticker ORDER BY date ROWS BETWEEN 19 PRECEDING AND CURRENT ROW) as ma20,
 AVG(close) OVER (PARTITION BY ticker ORDER BY date ROWS BETWEEN 49 PRECEDING AND CURRENT ROW) as ma50
 FROM (
 SELECT ticker, date, close
 FROM (
 UNPIVOT stocks ON columns INCLUDE (date) INTO name, value
 )
 WHERE name = 'close'
 )
 ),
 signals AS (
 SELECT ticker, date, close, ma20, ma50,
 LAG(ma20, 1) OVER (PARTITION BY ticker ORDER BY date) as ma20_prev,
 LAG(ma50, 1) OVER (PARTITION BY ticker ORDER BY date) as ma50_prev
 FROM daily
 )
 SELECT ticker, date,
 CASE
 WHEN ma20 > ma50 AND ma20_prev <= ma50_prev THEN '🟢 Golden Cross'
 WHEN ma20 < ma50 AND ma20_prev >= ma50_prev THEN '🔴 Death Cross'
 ELSE '⚪ No Signal'
 END as signal,
 round(close, 2) as price
 FROM signals
 WHERE date >= CURRENT_DATE - INTERVAL '5' DAY
 ORDER BY date DESC, ticker
""").fetchdf()

print("\n📉 Latest MA Signals:")
for _, row in ma_signals.iterrows():
 print(f" {row.ticker} ({row.date}): {row.signal} @ ${row.price}")

conn.close()

import duckdb
from datetime import datetime

conn = duckdb.connect("stocks.db")

report_date = datetime.now().strftime("%Y-%m-%d")
week_number = datetime.now().isocalendar()[1]

# 30-day gainers
gainers = conn.execute("""
 SELECT ticker,
 round(last(close) - first(close), 2) as change,
 round((last(close) / first(close) - 1) * 100, 2) as pct
 FROM (
 SELECT ticker, date, close
 FROM (UNPIVOT stocks ON columns INCLUDE (date) INTO name, value)
 WHERE name = 'close'
 )
 WHERE date >= CURRENT_DATE - INTERVAL '30' DAY
 GROUP BY ticker
 ORDER BY pct DESC
 LIMIT 5
""").fetchall()

conn.close()

# Assemble report
lines = [
 f"# 📊 Stock Weekly Report Week {week_number} ({report_date})",
 "",
 "## 🏆 Top 5 Gainers (30-day)",
 "",
]
for i, row in enumerate(gainers, 1):
 lines.append(f"{i}. **{row[0]}** — Change: {row[2]}% (Absolute: ${row[1]})")

lines += [
 "",
 "## 💡 Weekly Strategy Note",
 "",
 "> This content is for reference only and does not constitute investment advice. Investments involve risk.",
 "",
 f"*Report generated by DuckDB + yfinance | {report_date}*",
]

report = "\n".join(lines)

output_path = f"report_week{week_number}.md"
with open(output_path, "w") as f:
 f.write(report)
print(f"✅ Report generated: {output_path}")

# Edit cron tasks
crontab -e

# Add this line (5:00 PM UTC on Fridays = 1:00 AM Saturday Beijing time)
0 17 * * 5 cd /path/to/your/project && /usr/bin/python3 run.py

#!/usr/bin/env python3
"""Unified entry: fetch data → analyze → generate report"""
import subprocess
import sys

steps = [
 ("Fetching data", "data_fetcher.py"),
 ("Running analysis", "analyzer.py"),
 ("Generating report", "report_generator.py"),
]

for name, script in steps:
 print(f"\n{'='*40}")
 print(f"▶ {name}: {script}")
 print('='*40)
 result = subprocess.run([sys.executable, script], capture_output=True, text=True)
 print(result.stdout)
 if result.returncode != 0:
 print(f"❌ {name} failed: {result.stderr}")
 sys.exit(1)

print("\n🎉 All steps completed successfully!")

0 17 * * 5 cd /path/to/your/project && /usr/bin/python3 run.py

import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart

def send_report(email_to, report_path, smtp_server, smtp_port, smtp_user, smtp_pass):
 with open(report_path, "r") as f:
 body = f.read()

 msg = MIMEMultipart()
 msg['From'] = smtp_user
 msg['To'] = email_to
 msg['Subject'] = f"📊 Stock Weekly Report - {datetime.now().strftime('%Y-%m-%d')}"
 msg.attach(MIMEText(body, 'plain', 'utf-8'))

 server = smtplib.SMTP(smtp_server, smtp_port)
 server.starttls()
 server.login(smtp_user, smtp_pass)
 server.send_message(msg)
 server.quit()
 print("📧 Email sent")

import requests

def send_to_telegram(chat_id, bot_token, message):
 url = f"https://api.telegram.org/bot{bot_token}/sendMessage"
 requests.post(url, json={
 "chat_id": chat_id,
 "text": message,
 "parse_mode": "Markdown"
 })
 print("📱 Telegram message sent")