pip install duckdb akshare pandas

import akshare as ak
import duckdb
import pandas as pd
from datetime import datetime, timedelta

# 连接 DuckDB（自动创建本地文件数据库）
con = duckdb.connect("sector_monitor.db")

# 获取申万一级行业列表
sector_df = ak.stock_board_industry_name_em()
print(f"监测 {len(sector_df)} 个行业板块")

# 取最近 90 个自然日的数据（约 60 个交易日）
end_date = datetime.now().strftime("%Y%m%d")
start_date = (datetime.now() - timedelta(days=90)).strftime("%Y%m%d")

# 创建主表
con.execute("""
    CREATE TABLE IF NOT EXISTS sector_daily (
        日期 DATE,
        sector VARCHAR,
        开盘 DOUBLE,
        收盘 DOUBLE,
        最高 DOUBLE,
        最低 DOUBLE,
        成交量 BIGINT,
        成交额 BIGINT,
        振幅 DOUBLE,
        涨跌幅 DOUBLE,
        涨跌额 DOUBLE,
        换手率 DOUBLE
    )
""")

for _, row in sector_df.head(5).iterrows():  # 先测试前5个板块
    sector_name = row["板块名称"]
    try:
        df = ak.stock_board_industry_hist_em(
            symbol=sector_name,
            start_date=start_date,
            end_date=end_date,
            period="daily",
            adjust="qfq"
        )
        if df.empty:
            continue
        df["sector"] = sector_name
        
        # 将 DataFrame 注册为临时表并写入
        con.register("df_tmp", df)
        con.execute("""
            INSERT INTO sector_daily 
            SELECT 日期, sector, 开盘, 收盘, 最高, 最低, 
                   成交量, 成交额, 振幅, 涨跌幅, 涨跌额, 换手率
            FROM df_tmp
        """)
        print(f"  ✓ {sector_name}: {len(df)} 条记录")
    except Exception as e:
        print(f"  ✗ {sector_name}: {e}")

-- 建立板块动量视图
CREATE OR REPLACE VIEW sector_momentum AS
WITH daily_return AS (
    SELECT
        sector,
        日期,
        (收盘 - LAG(收盘) OVER (PARTITION BY sector ORDER BY 日期)) 
            / NULLIF(LAG(收盘) OVER (PARTITION BY sector ORDER BY 日期), 0) AS daily_ret
    FROM sector_daily
),
momentum AS (
    SELECT
        sector,
        MAX(日期) AS latest_date,
        -- 5日累计收益（使用几何累乘更精确）
        EXP(SUM(LN(1 + COALESCE(daily_ret, 0))) 
            OVER (PARTITION BY sector ORDER BY 日期 
                  ROWS BETWEEN 4 PRECEDING AND CURRENT ROW)) - 1 AS ret_5d,
        -- 20日累计收益
        EXP(SUM(LN(1 + COALESCE(daily_ret, 0))) 
            OVER (PARTITION BY sector ORDER BY 日期 
                  ROWS BETWEEN 19 PRECEDING AND CURRENT ROW)) - 1 AS ret_20d,
        -- 60日累计收益
        EXP(SUM(LN(1 + COALESCE(daily_ret, 0))) 
            OVER (PARTITION BY sector ORDER BY 日期 
                  ROWS BETWEEN 59 PRECEDING AND CURRENT ROW)) - 1 AS ret_60d,
        -- 20日平均成交额（判断资金活跃度）
        AVG(成交额) OVER (PARTITION BY sector ORDER BY 日期 
                          ROWS BETWEEN 19 PRECEDING AND CURRENT ROW) AS avg_volume_20d
    FROM daily_return
)
SELECT DISTINCT
    sector,
    ret_5d,
    ret_20d,
    ret_60d,
    avg_volume_20d,
    -- 动量综合评分：短期趋势权重最高
    ret_5d * 0.5 + ret_20d * 0.3 + ret_60d * 0.2 AS momentum_score
FROM momentum
WHERE 日期 = (SELECT MAX(日期) FROM daily_return)
ORDER BY momentum_score DESC;

-- 排名与信号生成
CREATE OR REPLACE VIEW sector_signals AS
WITH ranked AS (
    SELECT *,
        ROW_NUMBER() OVER (ORDER BY momentum_score DESC) AS rank_asc,
        ROW_NUMBER() OVER (ORDER BY momentum_score ASC) AS rank_desc
    FROM sector_momentum
)
SELECT
    sector,
    ROUND(ret_5d * 100, 2) AS ret_5d_pct,
    ROUND(ret_20d * 100, 2) AS ret_20d_pct,
    ROUND(momentum_score * 100, 2) AS score,
    ROUND(avg_volume_20d / 1e8, 1) AS avg_amount_yi,  -- 单位：亿元
    CASE 
        WHEN rank_asc <= 5 THEN '🔥 强势领涨'
        WHEN rank_asc <= 15 THEN '⚡ 动量偏强'
        WHEN rank_desc <= 5 THEN '❄️ 弱势回避'
        ELSE '➡️ 中性'
    END AS signal,
    CASE
        WHEN rank_asc <= 5 AND avg_volume_20d > 1e10 THEN '买入'  -- 强势 + 放量
        WHEN rank_desc <= 5 THEN '卖出/回避'
        ELSE '持有/观望'
    END AS action
FROM ranked
ORDER BY rank_asc;

# 用 DuckDB 直接生成报告文本
report = con.execute("""
SELECT 
    strftime(current_date, '%Y-%m-%d') || ' A股板块轮动日报' AS title,
    '---' AS sep1,
    '🔥 强势板块 Top 5:' AS section1,
    string_agg(
        '  ' || sector || ' | 5日: ' || ret_5d_pct || '% | 动量分: ' || score || ' | 信号: ' || action,
        chr(10)
    ) FILTER (WHERE rank_asc <= 5) AS top_sectors,
    '---' AS sep2,
    '❄️ 弱势板块 Top 5:' AS section2,
    string_agg(
        '  ' || sector || ' | 20日: ' || ret_20d_pct || '% | 信号: ' || action,
        chr(10)
    ) FILTER (WHERE rank_desc <= 5) AS bottom_sectors,
    '---' AS sep3,
    '💡 操作建议:' AS section3,
    CASE
        WHEN count(*) FILTER (WHERE action = '买入') >= 3 
        THEN '市场情绪偏乐观，关注强势板块回调后的二次入场机会'
        WHEN count(*) FILTER (WHERE action = '买入') = 0 
        THEN '无明确买入信号，建议观望或关注逆势抗跌板块'
        ELSE '结构性行情，关注动量持续居前的板块'
    END AS advice
FROM sector_signals
""").fetchone()

report_text = '\n'.join([str(r) for r in report if r])
print(report_text)

def send_telegram(bot_token, chat_id, text):
    import requests
    url = f"https://api.telegram.org/bot{bot_token}/sendMessage"
    requests.post(url, json={"chat_id": chat_id, "text": text})

#!/bin/bash
cd /path/to/project
python3 collect_data.py     # 数据采集
python3 compute_signals.py  # 指标计算
python3 send_report.py      # 推送报告

CREATE OR REPLACE VIEW multi_factor_score AS
SELECT 
    m.sector,
    m.momentum_score * 0.3 +      -- 动量因子
    v.volume_change * 0.2 +       -- 成交量变化因子
    p.price_stability * 0.2 +     -- 价格稳定性因子
    r.relative_strength * 0.3     -- 相对强度因子
    AS composite_score
FROM sector_momentum m
JOIN sector_volume v USING (sector)
JOIN sector_stability p USING (sector)
JOIN sector_rel_strength r USING (sector);

-- 回测：每周调仓，买入动量前 3 的板块
WITH weekly_rank AS (
    SELECT 
        日期,
        sector,
        momentum_score,
        ROW_NUMBER() OVER (PARTITION BY 日期 ORDER BY momentum_score DESC) AS rnk
    FROM sector_daily_momentum
    WHERE dayofweek(日期) = 5  -- 每周五调仓
)
SELECT 
    sector,
    COUNT(*) AS hold_weeks,
    AVG(ret_20d) AS avg_return,
    STDDEV(ret_20d) AS volatility,
    AVG(ret_20d) / NULLIF(STDDEV(ret_20d), 0) AS sharpe_ratio
FROM weekly_rank
WHERE rnk <= 3
GROUP BY sector
ORDER BY sharpe_ratio DESC;

价格方案：
- 基础版（99 元/月）：每日板块轮动推送 + Top/Bottom 5
- 专业版（299 元/月）：基础版 + 多因子评分 + 个股筛选
- 企业版（999 元/月）：专业版 + 历史回测报告 + 定制因子