用 DuckDB 搭建自动化持仓看板：30 行代码实现每日盈亏推送

pip install duckdb yfinance pandas matplotlib

pip install duckdb yfinance pandas matplotlib -i https://pypi.tuna.tsinghua.edu.cn/simple

import duckdb
import yfinance as yf
import pandas as pd
from datetime import datetime, timedelta

# ── 持仓定义（你的真实持仓）─────────────────────────
portfolio = pd.DataFrame({
    'ticker': ['AAPL', 'MSFT', 'GOOGL', 'NVDA', 'TSLA'],
    'shares': [100, 50, 20, 30, 40],
    'cost_basis': [145.50, 320.00, 138.20, 850.00, 245.00],  # 买入均价
})

# ── 拉取实时行情 ───────────────────────────────────
tickers = portfolio['ticker'].tolist()
today = datetime.now().strftime('%Y-%m-%d')
current_prices = {}

for t in tickers:
    try:
        stock = yf.Ticker(t)
        hist = stock.history(period='1d')
        if not hist.empty:
            current_prices[t] = hist['Close'].iloc[-1]
    except Exception as e:
        print(f"⚠ {t} 获取失败: {e}")
        current_prices[t] = None

print(f"✅ 获取到 {len([p for p in current_prices.values() if p])}/{len(tickers)} 只股票行情")

# ── 写入 DuckDB 内存数据库（零文件写入）────────────
con = duckdb.connect(':memory:')

# 持仓表
con.execute("CREATE TABLE portfolio AS SELECT * FROM portfolio")

# 行情表
price_rows = [
    (t, p, today) for t, p in current_prices.items() if p
]
con.execute("CREATE TABLE prices AS SELECT * FROM (VALUES ?)", [price_rows])

# 验证数据
print(con.execute("SELECT * FROM portfolio").fetchdf())
print(con.execute("SELECT * FROM prices").fetchdf())

# ── 核心查询：一次 SQL 搞定所有计算 ────────────────
profit_query = """
SELECT
    p.ticker,
    p.shares,
    p.cost_basis,
    pr.close AS current_price,
    (pr.close - p.cost_basis) * p.shares AS profit_loss,
    ROUND(((pr.close - p.cost_basis) / p.cost_basis) * 100, 2) AS pct_change,
    pr.close * p.shares AS market_value
FROM portfolio p
JOIN prices pr ON p.ticker = pr.ticker
ORDER BY profit_loss DESC
"""

result = con.execute(profit_query).fetchdf()

print("=" * 50)
print(f"📊 持仓日报 | {today}")
print("=" * 50)
for _, row in result.iterrows():
    emoji = "📈" if row['profit_loss'] > 0 else "📉"
    sign = "+" if row['profit_loss'] > 0 else ""
    print(f"{emoji} {row['ticker']:6s} | "
          f"盈亏: {sign}{row['profit_loss']:,.2f} | "
          f"涨幅: {sign}{row['pct_change']}% | "
          f"市值: ${row['market_value']:,.2f}")

total_pl = result['profit_loss'].sum()
total_mv = result['market_value'].sum()
print("-" * 50)
print(f"💰 总盈亏: {total_pl:+,.2f} | 总市值: ${total_mv:,.2f}")
print("=" * 50)

con.close()

import matplotlib.pyplot as plt
plt.rcParams['font.sans-serif'] = ['SimHei', 'DejaVu Sans']

fig, axes = plt.subplots(1, 2, figsize=(12, 5))

# 左图：盈亏柱状图
colors = ['#2ecc71' if x > 0 else '#e74c3c' for x in result['profit_loss']]
axes[0].bar(result['ticker'], result['profit_loss'], color=colors)
axes[0].set_title('Daily P&L by Stock', fontsize=12)
axes[0].axhline(y=0, color='black', linewidth=0.5)
axes[0].tick_params(axis='x', rotation=45)

# 右图：市值占比
axes[1].pie(result['market_value'], labels=result['ticker'], autopct='%1.1f%%')
axes[1].set_title('Portfolio Allocation', fontsize=12)

plt.tight_layout()
plt.savefig('daily_report.png', dpi=150, bbox_inches='tight')
plt.close()
print("✅ 图表已保存: daily_report.png")

import schedule
import time
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from email.mime.image import MIMEImage

def send_report():
    """推送日报到微信/邮件"""
    smtp_server = "smtp.gmail.com"
    smtp_port = 587
    sender = "[email protected]"
    password = "your_app_password"
    receivers = ["[email protected]"]
    
    msg = MIMEMultipart()
    msg['From'] = sender
    msg['To'] = ", ".join(receivers)
    msg['Subject'] = f"📊 持仓日报 {today}"
    
    body = f"持仓日报 {today}\n{'='*40}\n"
    for _, row in result.iterrows():
        sign = "+" if row['profit_loss'] > 0 else ""
        body += f"{row['ticker']}: {sign}{row['profit_loss']:,.2f} ({sign}{row['pct_change']}%)\n"
    body += f"\n总盈亏: {total_pl:+,.2f}\n总市值: ${total_mv:,.2f}"
    
    msg.attach(MIMEText(body, 'plain', 'utf-8'))
    
    # 附加图表
    with open('daily_report.png', 'rb') as f:
        img = MIMEImage(f.read())
        img.add_header('Content-Disposition', 'attachment', filename='daily_report.png')
        msg.attach(img)
    
    server = smtplib.SMTP(smtp_server, smtp_port)
    server.starttls()
    server.login(sender, password)
    server.sendmail(sender, receivers, msg.as_string())
    server.quit()
    
    print(f"✅ 日报已发送至 {len(receivers)} 个收件人")

# ── 定时任务：每天 22:00 执行 ─────────────────────
schedule.every().day.at("22:00").do(send_report)

print("🤖 自动推送服务已启动，等待 22:00...")
while True:
    schedule.run_pending()
    time.sleep(60)

# 编辑 crontab
crontab -e

# 每天 22:00 自动运行
0 22 * * * cd /path/to/duckdb-portfolio && python3 reporter.py >> /var/log/portfolio.log 2>&1

# 1. 支持多持仓组合
portfolios = {
    '成长型': {'AAPL': 100, 'NVDA': 30, 'TSLA': 40},
    '稳健型': {'MSFT': 50, 'GOOGL': 20, 'JNJ': 60},
}

# 2. 支持涨跌预警
ALERT_THRESHOLD = 5.0  # 涨跌幅超过 5% 发送预警
if abs(row['pct_change']) > ALERT_THRESHOLD:
    send_alert(f"⚠️ {row['ticker']} 涨跌幅超过 {ALERT_THRESHOLD}%!")

# 3. 支持 Telegram/微信推送（替代邮件）
import requests
TELEGRAM_BOT_TOKEN = "your_bot_token"
TELEGRAM_CHAT_ID = "your_chat_id"

def send_telegram(msg):
    url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage"
    requests.post(url, json={
        'chat_id': TELEGRAM_CHAT_ID,
        'text': msg,
        'parse_mode': 'HTML'
    })