A_Share_DP/src/ashare_dp/market/recommendations.py

"""Stock Recommendation Engine — money flow driven stock picks.

Identifies top stocks in sectors with strong capital inflow,
ranks by momentum/volume/trend, generates actionable trade plans
with entry/stop/target levels.
"""

from __future__ import annotations

from datetime import date, datetime

import duckdb
from loguru import logger

from ashare_dp.data.store.database import analytics_conn, kline_glob


def get_recommendations(
    trade_date: date,
    top_n: int = 10,
    min_amount_yi: float = 2.0,  # 最低日成交额（亿）
) -> list[dict]:
    """Generate ranked stock recommendations based on money flow.

    1. Find sectors with strongest capital inflow
    2. Within each sector, rank stocks by momentum + volume + trend
    3. Generate trade plans (entry, stop, targets)

    Returns list of recommendation dicts sorted by composite score.
    """
    parquet_glob = kline_glob()
    conn = analytics_conn()

    # ═══ Step 1: Find top inflow sectors ═══
    try:
        flow_sql = f"""
            WITH normalized AS (
                SELECT ts_code, trade_date, amount
                FROM read_parquet('{parquet_glob}', hive_partitioning=true, union_by_name=true)
                WHERE trade_date >= $start_date AND trade_date <= $end_date
            ),
            daily AS (
                SELECT n.trade_date, n.amount, si.industry_name
                FROM normalized n
                JOIN stock_industry si ON n.ts_code = si.ts_code
            ),
            industry_daily AS (
                SELECT industry_name, trade_date, SUM(amount) AS total_amount
                FROM daily GROUP BY industry_name, trade_date
            ),
            ranked AS (
                SELECT *, ROW_NUMBER() OVER (PARTITION BY industry_name ORDER BY trade_date DESC) AS rn
                FROM industry_daily
            ),
            recent AS (
                SELECT industry_name, SUM(total_amount) AS recent
                FROM ranked WHERE rn <= 5 GROUP BY industry_name
            ),
            prior AS (
                SELECT industry_name, SUM(total_amount) AS prior
                FROM ranked WHERE rn > 5 AND rn <= 10 GROUP BY industry_name
            )
            SELECT COALESCE(r.industry_name, p.industry_name) AS industry_name,
                   CASE WHEN COALESCE(p.prior,0) > 0 THEN (COALESCE(r.recent,0) - p.prior) / p.prior ELSE 0 END AS flow_change
            FROM recent r FULL OUTER JOIN prior p ON r.industry_name = p.industry_name
            WHERE COALESCE(r.recent, 0) + COALESCE(p.prior, 0) > 0
            ORDER BY flow_change DESC
            LIMIT 5
        """
        end_str = trade_date.strftime("%Y-%m-%d")
        from datetime import timedelta
        start_str = (trade_date - timedelta(days=20)).strftime("%Y-%m-%d")
        flow_df = conn.execute(
            flow_sql.replace("$start_date", f"'{start_str}'").replace("$end_date", f"'{end_str}'")
        ).fetchdf()
        top_sectors = flow_df["industry_name"].tolist() if not flow_df.empty else []
    except Exception as e:
        logger.warning(f"Flow sector query failed: {e}")
        top_sectors = []
    finally:
        conn.close()

    if not top_sectors:
        return []

    # ═══ Step 2: Find best stocks in top sectors ═══
    conn = analytics_conn()
    recommendations = []

    for sector in top_sectors[:3]:  # top 3 inflow sectors
        try:
            sector_list = "', '".join(top_sectors)
            stock_sql = f"""
                WITH normalized AS (
                    SELECT ts_code, trade_date, trade_time, close, volume, amount, high, low, open
                    FROM read_parquet('{parquet_glob}', hive_partitioning=true, union_by_name=true)
                    WHERE trade_date <= $trade_date
                ),
                with_ma AS (
                    SELECT *,
                        AVG(close) OVER (PARTITION BY ts_code ORDER BY trade_time ROWS BETWEEN 19 PRECEDING AND CURRENT ROW) AS ma20,
                        AVG(close) OVER (PARTITION BY ts_code ORDER BY trade_time ROWS BETWEEN 59 PRECEDING AND CURRENT ROW) AS ma60,
                        AVG(volume) OVER (PARTITION BY ts_code ORDER BY trade_time ROWS BETWEEN 19 PRECEDING AND CURRENT ROW) AS vol_ma20,
                        (high - low) AS day_range,
                        ROW_NUMBER() OVER (PARTITION BY ts_code ORDER BY trade_time DESC) AS rn
                    FROM normalized
                ),
                with_atr AS (
                    SELECT *,
                        AVG(day_range) OVER (PARTITION BY ts_code ORDER BY trade_time ROWS BETWEEN 19 PRECEDING AND CURRENT ROW) AS atr20
                    FROM with_ma
                ),
                latest AS (
                    SELECT w.*, si.industry_name
                    FROM with_atr w
                    JOIN stock_industry si ON w.ts_code = si.ts_code
                    WHERE w.rn = 1 AND si.industry_name = $sector
                ),
                prev AS (
                    SELECT ts_code, close AS close_prev
                    FROM with_ma WHERE rn = 6
                ),
                scored AS (
                    SELECT
                        l.ts_code, l.industry_name, l.close, l.volume, l.amount,
                        l.ma20, l.ma60, l.vol_ma20, l.atr20,
                        l.high, l.low, l.open,
                        p.close_prev,
                        -- Momentum: 5d return
                        (l.close - p.close_prev) / NULLIF(p.close_prev, 0) AS ret_5d,
                        -- Volume expansion
                        l.volume / NULLIF(l.vol_ma20, 0) AS vol_ratio,
                        -- Trend: above MAs
                        CASE WHEN l.close > l.ma20 THEN 1 ELSE 0 END + CASE WHEN l.close > l.ma60 THEN 1 ELSE 0 END AS trend_score
                    FROM latest l
                    LEFT JOIN prev p ON l.ts_code = p.ts_code
                    WHERE l.amount > $min_amt
                )
                SELECT *,
                    (COALESCE(ret_5d, 0) * 50 + LEAST(vol_ratio, 3.0) / 3.0 * 25 + trend_score * 12.5) AS composite
                FROM scored
                ORDER BY composite DESC
                LIMIT 4
            """
            min_amt = min_amount_yi * 1e8
            stock_df = conn.execute(
                stock_sql,
                {"trade_date": trade_date.isoformat(), "sector": sector, "min_amt": min_amt},
            ).fetchdf()

            # Look up stock names
            stock_names = {}
            try:
                nc = duckdb.connect("data/duckdb/ashare.db", read_only=True)
                for c in stock_df["ts_code"].tolist():
                    r = nc.execute("SELECT name FROM stock_info WHERE ts_code=?", [c]).fetchone()
                    stock_names[c] = r[0] if r else c
                nc.close()
            except Exception:
                pass

            for _, row in stock_df.iterrows():
                import math
                entry = float(row["close"])
                atr_raw = row.get("atr20")
                atr = float(atr_raw) if atr_raw and not (isinstance(atr_raw, float) and math.isnan(atr_raw)) else entry * 0.03

                # Stop: 2.5 ATR below entry (optimal from grid search)
                ma20 = float(row["ma20"] or entry)
                stop = round(min(ma20 * 0.97, entry - 2.5 * atr), 2)

                # Targets: 3.0 ATR (optimal), secondary 4.0 ATR
                target1 = round(entry + 3.0 * atr, 2)
                target2 = round(entry + 4.0 * atr, 2)

                # Risk/reward
                risk = entry - stop
                reward = target1 - entry
                rr_ratio = round(reward / risk, 1) if risk > 0 else 0

                # Position sizing guidance
                position_advice = "Standard"
                if float(row["trend_score"]) >= 2:
                    position_advice = "Aggressive"
                elif float(row.get("vol_ratio", 1)) < 0.8:
                    position_advice = "Reduced"

                import math as _m
                def _safe(v, d):
                    try:
                        f = float(v)
                        return d if _m.isnan(f) else f
                    except (ValueError, TypeError):
                        return d
                _ret5 = _safe(row.get("ret_5d"), 0.0)
                _volr = _safe(row.get("vol_ratio"), 1.0)
                _trnd = _safe(row.get("trend_score"), 0.0)
                _comp = _safe(row.get("composite"), 0.0)
                if _m.isnan(_ret5): _ret5 = 0.0
                if _m.isnan(_volr): _volr = 1.0
                if _m.isnan(_trnd): _trnd = 0.0
                if _m.isnan(_comp): _comp = 0.0

                ts = row["ts_code"]
                recommendations.append({
                    "ts_code": ts,
                    "name": stock_names.get(ts, ts),
                    "sector": sector,
                    "entry": round(entry, 2),
                    "stop": stop,
                    "target1": target1,
                    "target2": target2,
                    "rr_ratio": round(float(rr_ratio), 1) if not _m.isnan(rr_ratio) else 0.0,
                    "position": position_advice,
                    "score": round(_comp, 1),
                    "metrics": {
                        "ret_5d": round(_ret5 * 100, 1),
                        "vol_ratio": round(_volr, 2),
                        "trend": "Strong" if _trnd >= 2 else "Weak",
                        "amount_yi": round(float(row["amount"]) / 1e8, 1),
                    },
                    "action": _recommend_action(_ret5, _volr, _trnd),
                })

        except Exception as e:
            logger.warning(f"Stock scoring failed for sector {sector}: {e}")

    conn.close()

    # Sort by composite score
    recommendations.sort(key=lambda r: r["score"], reverse=True)
    return recommendations[:top_n]


def _recommend_action(ret_5d: float, vol_ratio: float, trend_score: float) -> str:
    """Generate trading action recommendation."""
    if trend_score >= 2 and ret_5d > 0.03 and vol_ratio > 1.2:
        return "买入 — 趋势强势，放量上涨"
    elif trend_score >= 2 and ret_5d > 0:
        return "关注 — 趋势健康，等待回踩"
    elif trend_score >= 1 and vol_ratio > 1.0:
        return "观察 — 趋势形成中，可轻仓试"
    elif ret_5d < -0.03:
        return "回避 — 短期偏弱，等待企稳"
    else:
        return "观望 — 方向不明，暂不参与"