import duckdb
import pandas as pd

# 创建 DuckDB 数据库文件（持久化存储）
con = duckdb.connect("stock_data.duckdb")

# 财务数据表：存储每日更新的财务指标
con.execute("""
CREATE TABLE IF NOT EXISTS financials (
    ticker VARCHAR,
    date DATE,
    revenue BIGINT,
    net_income BIGINT,
    total_assets BIGINT,
    total_debt BIGINT,
    eps REAL,
    pe_ratio REAL
)
""")

# 每日快照表：用于时间序列分析
con.execute("""
CREATE TABLE IF NOT EXISTS daily_snapshots (
    ticker VARCHAR,
    snapshot_date DATE,
    market_cap BIGINT,
    volume BIGINT,
    price REAL
)
""")

# 创建索引加速查询
con.execute("CREATE INDEX IF NOT EXISTS idx_financials_ticker_date ON financials(ticker, date)")
con.execute("CREATE INDEX IF NOT EXISTS idx_snapshots_ticker_date ON daily_snapshots(ticker, snapshot_date)")

print("✅ 数据模型创建完成")

import numpy as np
from datetime import datetime, timedelta

# 模拟最近30天的财务数据
np.random.seed(42)
tickers = ['AAPL', 'MSFT', 'GOOGL', 'AMZN', 'TSLA', 'NVDA', 'META', 'JPM']

data_rows = []
snapshot_rows = []
base_date = datetime(2026, 8, 9)

for i, ticker in enumerate(tickers):
    for day in range(30):
        date = base_date - timedelta(days=day)
        revenue = int(50e9 + np.random.randn() * 5e9 + i * 10e9)
        net_income = int(revenue * 0.25 + np.random.randn() * 2e9)
        eps = round(2.0 + np.random.randn() * 0.5 + i * 0.3, 2)
        pe_ratio = round(20 + np.random.randn() * 8 + i * 2, 2)
        
        data_rows.append((ticker, date, revenue, net_income, 
                          int(revenue * 8), int(revenue * 0.4), eps, pe_ratio))
        
        snapshot_rows.append((
            ticker, date,
            int(revenue * 15 + np.random.randn() * 1e9),
            int(50e6 + np.random.randn() * 10e6),
            round(pe_ratio * eps, 2)
        ))

# 批量插入
con.execute("""
    INSERT INTO financials 
    VALUES (SELECT * FROM (VALUES ?) AS t(ticker, date, revenue, net_income, total_assets, total_debt, eps, pe_ratio))
""", [tuple(row) for row in data_rows])

con.execute("""
    INSERT INTO daily_snapshots 
    VALUES (SELECT * FROM (VALUES ?) AS t(ticker, snapshot_date, market_cap, volume, price))
""", [tuple(row) for row in snapshot_rows])

print(f"✅ 入库完成：{len(data_rows)} 条财务记录，{len(snapshot_rows)} 条快照记录")

signal_sql = """
WITH ranked_stocks AS (
    SELECT 
        ticker,
        date,
        -- 净资产收益率（近12个月滚动）
        AVG(net_income * 4.0 / NULLIF(total_assets, 0)) AS avg_roe,
        -- 收入同比增长率（最近5日 vs 之前5日）
        (
            AVG(CASE WHEN date >= date - INTERVAL '5' DAY THEN revenue END) /
            NULLIF(AVG(CASE WHEN date < date - INTERVAL '5' DAY THEN revenue END), 0) - 1
        ) AS revenue_growth,
        -- 负债率
        AVG(total_debt * 1.0 / NULLIF(total_assets, 0)) AS debt_ratio,
        -- PE分位数
        PERCENT_RANK() OVER (
            PARTITION BY date ORDER BY pe_ratio
        ) AS pe_percentile,
        -- 综合得分
        (
            AVG(avg_roe) * 0.4 
            + COALESCE(revenue_growth, 0) * 0.3 
            + (1 - AVG(debt_ratio)) * 0.2
            + (1 - PERCENT_RANK() OVER (PARTITION BY date ORDER BY pe_ratio)) * 0.1
        ) AS composite_score
    FROM financials
    WHERE date >= date - INTERVAL '30' DAY
    GROUP BY ticker, date
    HAVING avg_roe > 0.10          -- ROE > 10%
       AND debt_ratio < 0.6        -- 负债率 < 60%
       AND pe_ratio < 50           -- PE < 50
)
SELECT * FROM ranked_stocks
ORDER BY date DESC, composite_score DESC
LIMIT 20
"""

signals = con.execute(signal_sql).fetchdf()
print(signals.to_string())

from fastapi import FastAPI
from pydantic import BaseModel
import duckdb

app = FastAPI(title="DuckDB 选股数据 API")

# 进程内单例连接（避免每次请求重新打开文件）
_db_path = "stock_data.duckdb"

class SignalRequest(BaseModel):
    date: str
    top_n: int = 10

@app.get("/api/signals")
def get_signals(req: SignalRequest):
    con = duckdb.connect(_db_path)
    result = con.execute(f"""
        WITH ranked_stocks AS (
            SELECT 
                ticker,
                AVG(net_income * 4.0 / NULLIF(total_assets, 0)) AS avg_roe,
                (
                    AVG(CASE WHEN date >= '{req.date}' - INTERVAL '5' DAY THEN revenue END) /
                    NULLIF(AVG(CASE WHEN date < '{req.date}' - INTERVAL '5' DAY THEN revenue END), 0) - 1
                ) AS revenue_growth,
                AVG(total_debt * 1.0 / NULLIF(total_assets, 0)) AS debt_ratio,
                (
                    AVG(avg_roe) * 0.4 
                    + COALESCE(revenue_growth, 0) * 0.3 
                    + (1 - AVG(debt_ratio)) * 0.2
                    + (1 - PERCENT_RANK() OVER (PARTITION BY date ORDER BY pe_ratio)) * 0.1
                ) AS composite_score
            FROM financials
            WHERE date = '{req.date}'
            GROUP BY ticker
            HAVING avg_roe > 0.10 AND debt_ratio < 0.6 AND pe_ratio < 50
        )
        SELECT ticker, avg_roe, revenue_growth, debt_ratio, composite_score
        FROM ranked_stocks
        ORDER BY composite_score DESC
        LIMIT {req.top_n}
    """).fetchdf()
    return result.to_dict(orient='records')

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

# 获取今日 Top 10 选股信号
curl -X POST "http://localhost:8000/api/signals" \
  -H "Content-Type: application/json" \
  -d '{"date": "2026-08-09", "top_n": 10}'

# 返回结果
[
  {"ticker": "NVDA", "avg_roe": 0.324, "revenue_growth": 0.152, "debt_ratio": 0.213, "composite_score": 0.487},
  {"ticker": "META", "avg_roe": 0.288, "revenue_growth": 0.099, "debt_ratio": 0.188, "composite_score": 0.456},
  ...
]

# 每天收盘后 17:00 自动更新数据
0 17 * * * cd /home/user/stock-screener && python3 update_data.py >> /var/log/stock_update.log 2>&1

import duckdb
import requests
from datetime import datetime

con = duckdb.connect("stock_data.duckdb")

# 1. 拉取最新数据（对接 Yahoo Finance 或 Alpha Vantage）
# 2. 增量更新到 DuckDB
# 3. 重新计算选股信号
# 4. 发送通知（Telegram/微信）