diff --git a/docs/superpowers/specs/2026-04-07-multi-timeframe-analysis-design.md b/docs/superpowers/specs/2026-04-07-multi-timeframe-analysis-design.md new file mode 100644 index 0000000..440f8be --- /dev/null +++ b/docs/superpowers/specs/2026-04-07-multi-timeframe-analysis-design.md @@ -0,0 +1,120 @@ +# Multi-Timeframe Technical Analysis + +## Context + +The current `analyze technical` command only operates on daily data, limiting the user's ability to perceive medium-to-long-term trends. Weekly and monthly timeframe indicators provide essential context for understanding whether a stock is in an uptrend, downtrend, or consolidation at a macro level. This feature adds weekly/monthly technical indicator output to complement the existing daily analysis — the CLI provides data only; qualitative interpretation is left to the agent. + +## Design Principle + +**CLI = data provider, Agent = interpreter.** The CLI outputs numerical indicator values for each timeframe without making qualitative judgments (no "bullish"/"bearish" labels). The Claude skill layer interprets the data. + +## Approach: Computation-Layer Aggregation + +Aggregate existing daily price bars into weekly/monthly bars at computation time. No changes to `PriceBar` model, database schema, or market providers. + +**Why not API-based weekly/monthly data?** +- Aggregated results are mathematically identical for complete periods +- CoinGecko doesn't support weekly/monthly K-lines, requiring a fallback anyway +- Zero additional API calls; all markets handled uniformly + +## Data Aggregation + +### New file: `haoinvest/engine/aggregation.py` + +Two public functions: + +```python +def aggregate_to_weekly(daily_bars: list[PriceBar]) -> list[PriceBar] +def aggregate_to_monthly(daily_bars: list[PriceBar]) -> list[PriceBar] +``` + +**Aggregation rules per period (week/month):** + +| Field | Rule | +|-------|------| +| `open` | First trading day's open | +| `high` | max(all highs) | +| `low` | min(all lows) | +| `close` | Last trading day's close | +| `volume` | sum(all volumes) | +| `trade_date` | Last trading day's date | +| `symbol` | Inherited from source bars | +| `market_type` | Inherited from source bars | + +**Week boundary:** ISO week (Monday–Sunday). +**Month boundary:** Calendar month. + +**Handling incomplete periods:** The current (most recent) week/month may be incomplete. Include it as-is — the agent can account for this. + +### Data period requirement + +To compute monthly MA20 + MACD(26), need ~26 months of monthly data → ~2 years of daily data. The `analyze technical` default fetch period expands from 1 year to 2 years. + +## Technical Indicator Calculation + +Reuse existing `compute_technical()` from `haoinvest/engine/technical_engine.py`. It accepts a DataFrame and is timeframe-agnostic. + +**Indicators computed per timeframe (weekly & monthly):** +- MA: SMA(5, 10, 20), EMA(12, 26) +- MACD: DIF, DEA, histogram (fast=12, slow=26, signal=9) +- RSI(14) + +Bollinger Bands are excluded from weekly/monthly to keep output concise. + +**Call flow:** +1. Fetch 2 years of daily bars +2. `aggregate_to_weekly(bars)` → weekly bars → `pricebars_to_dataframe()` → `compute_technical()` → weekly indicators +3. `aggregate_to_monthly(bars)` → monthly bars → `pricebars_to_dataframe()` → `compute_technical()` → monthly indicators +4. Daily bars → existing flow unchanged + +## Output Structure + +### Text output (default) + +Layered from macro to micro: + +``` +📊 月线技术指标 (Monthly) + MA: MA5=xx.xx MA10=xx.xx MA20=xx.xx | EMA12=xx.xx EMA26=xx.xx + MACD: DIF=xx.xx DEA=xx.xx MACD=xx.xx + RSI(14): xx.xx + +📊 周线技术指标 (Weekly) + MA: MA5=xx.xx MA10=xx.xx MA20=xx.xx | EMA12=xx.xx EMA26=xx.xx + MACD: DIF=xx.xx DEA=xx.xx MACD=xx.xx + RSI(14): xx.xx + +📊 日线技术指标 (Daily) + [existing output unchanged] +``` + +### JSON output (`--format json`) + +Add `weekly` and `monthly` keys at the same level as the existing daily indicators: + +```json +{ + "monthly": { "ma": {...}, "macd": {...}, "rsi": ... }, + "weekly": { "ma": {...}, "macd": {...}, "rsi": ... }, + "daily": { ... } +} +``` + +## Files to Modify + +| File | Change | +|------|--------| +| `haoinvest/engine/aggregation.py` | **New** — weekly/monthly aggregation functions | +| `haoinvest/analysis/technical.py` | Add multi-timeframe analysis entry point | +| `haoinvest/cli/analyze.py` | Expand `technical` command to output weekly/monthly layers | +| `haoinvest/cli/formatters.py` | Format weekly/monthly indicator output | +| `haoinvest/models.py` | Add `timeframe` field to `TechnicalIndicators` (optional) | +| `tests/test_aggregation.py` | **New** — unit tests for aggregation logic | +| `tests/test_technical.py` | Add multi-timeframe analysis tests | + +## Verification + +1. **Unit tests:** Test aggregation with known daily data → verify weekly/monthly OHLCV values +2. **Integration test:** Run `haoinvest analyze technical 600519` → verify monthly/weekly/daily sections all appear +3. **JSON output:** Run with `--format json` → verify `monthly` and `weekly` keys present with correct structure +4. **Edge cases:** Test with < 1 month of data (monthly section should gracefully show what's available or indicate insufficient data) diff --git a/haoinvest/analysis/__init__.py b/haoinvest/analysis/__init__.py index 97cb5de..77c868e 100644 --- a/haoinvest/analysis/__init__.py +++ b/haoinvest/analysis/__init__.py @@ -3,13 +3,14 @@ from .fundamental import analyze_stock from .risk import calculate_risk_metrics, portfolio_correlation from .signals import aggregate_signals -from .technical import analyze_technical +from .technical import analyze_technical, analyze_technical_multi from .volume import analyze_volume __all__ = [ "aggregate_signals", "analyze_stock", "analyze_technical", + "analyze_technical_multi", "analyze_volume", "calculate_risk_metrics", "portfolio_correlation", diff --git a/haoinvest/analysis/technical.py b/haoinvest/analysis/technical.py index b9cd36a..9c38e9d 100644 --- a/haoinvest/analysis/technical.py +++ b/haoinvest/analysis/technical.py @@ -3,9 +3,10 @@ from datetime import date from ..db import Database +from ..engine.aggregation import aggregate_to_monthly, aggregate_to_weekly from ..engine.databridge import pricebars_to_dataframe from ..engine.technical_engine import compute_technical -from ..models import MarketType, TechnicalIndicators +from ..models import MarketType, MultiTimeframeTechnical, TechnicalIndicators def analyze_technical( @@ -37,3 +38,43 @@ def analyze_technical( result.symbol = symbol result.market_type = mt_str return result + + +def analyze_technical_multi( + db: Database, + symbol: str, + market_type: MarketType, + start_date: date | None = None, + end_date: date | None = None, + verbose: bool = False, +) -> MultiTimeframeTechnical: + """Calculate technical indicators across daily, weekly, and monthly timeframes.""" + bars = db.get_prices(symbol, market_type, start_date, end_date) + mt_str = market_type.value + + def _compute_for_bars(bar_list: list, timeframe: str) -> TechnicalIndicators: + df = pricebars_to_dataframe(bar_list) + if len(df) < 14: + return TechnicalIndicators( + symbol=symbol, + market_type=mt_str, + timeframe=timeframe, + message=f"Not enough {timeframe} data ({len(df)} bars, need at least 14)", + ) + result = compute_technical(df, verbose=verbose) + result.symbol = symbol + result.market_type = mt_str + result.timeframe = timeframe + return result + + daily = _compute_for_bars(bars, "daily") + weekly = _compute_for_bars(aggregate_to_weekly(bars), "weekly") + monthly = _compute_for_bars(aggregate_to_monthly(bars), "monthly") + + return MultiTimeframeTechnical( + symbol=symbol, + market_type=mt_str, + daily=daily, + weekly=weekly, + monthly=monthly, + ) diff --git a/haoinvest/cli/analyze.py b/haoinvest/cli/analyze.py index e1fb9ef..a1b39ad 100644 --- a/haoinvest/cli/analyze.py +++ b/haoinvest/cli/analyze.py @@ -8,11 +8,17 @@ from ..analysis.fundamental import analyze_stock from ..analysis.risk import calculate_risk_metrics, portfolio_correlation from ..analysis.signals import aggregate_signals -from ..analysis.technical import analyze_technical +from ..analysis.technical import analyze_technical, analyze_technical_multi from ..analysis.volume import analyze_volume from ..db import Database from ..models import MarketType -from .formatters import error_output, json_output, kv_output, tsv_output +from .formatters import ( + error_output, + json_output, + kv_output, + timeframe_section, + tsv_output, +) from .market import _detect_market_type app = typer.Typer(help="Analysis — fundamental, risk, technical, volume, signals.") @@ -32,6 +38,15 @@ def _ensure_prices_cached( existing = db.get_prices(symbol, market_type, start, end) if len(existing) > 10: + # Check if cached data covers the requested start date. + # If not, fetch the missing earlier portion. + earliest_cached = min(b.trade_date for b in existing) + if earliest_cached <= start + timedelta(days=7): + return + provider = get_provider(market_type) + bars = provider.get_price_history(symbol, start, earliest_cached) + if bars: + db.save_prices(bars) return provider = get_provider(market_type) bars = provider.get_price_history(symbol, start, end) @@ -254,23 +269,32 @@ def technical( ), use_json: bool = typer.Option(False, "--json", help="Output as JSON"), ) -> None: - """Technical indicators — MA, MACD, RSI, Bollinger Bands.""" + """Technical indicators — MA, MACD, RSI, Bollinger Bands (daily/weekly/monthly).""" db = _init_db() end_date = date.fromisoformat(end) if end else date.today() - start_date = date.fromisoformat(start) if start else end_date - timedelta(days=365) + start_date = date.fromisoformat(start) if start else end_date - timedelta(days=1095) symbol_list = [s.strip() for s in symbol.split(",")] if len(symbol_list) == 1: - # Single symbol — detailed kv_output (original behavior) + # Single symbol — multi-timeframe output mt = MarketType(market_type) if market_type else _detect_market_type(symbol) _ensure_prices_cached(db, symbol, mt, start_date, end_date) - result = analyze_technical( + multi = analyze_technical_multi( db, symbol, mt, start_date, end_date, verbose=verbose ) if use_json: - json_output(result) + json_output(multi) else: + # Monthly and weekly: compact timeframe sections + if multi.monthly: + timeframe_section("月线技术指标 (Monthly)", multi.monthly, verbose) + if multi.weekly: + timeframe_section("周线技术指标 (Weekly)", multi.weekly, verbose) + + # Daily: preserve original kv_output format for backward compatibility + result = multi.daily + print() # blank line before daily section output = {"symbol": result.symbol, "date": str(result.latest_date)} if result.message: output["message"] = result.message diff --git a/haoinvest/cli/formatters.py b/haoinvest/cli/formatters.py index c9754b8..07e87e1 100644 --- a/haoinvest/cli/formatters.py +++ b/haoinvest/cli/formatters.py @@ -56,6 +56,34 @@ def json_output(data: Any) -> None: print(json.dumps(data, ensure_ascii=False, indent=2, default=str)) +def timeframe_section( + label: str, + result: Any, + verbose: bool = False, +) -> None: + """Print a compact technical indicator section for a single timeframe.""" + print(f"\n📊 {label}") + if result.message: + print(f" {result.message}") + return + ma = result.moving_averages + print( + f" MA: SMA5={ma.sma_5} SMA10={ma.sma_10} SMA20={ma.sma_20}" + f" | EMA12={ma.ema_12} EMA26={ma.ema_26}" + ) + if verbose and ma.explanation: + print(f" MA说明: {ma.explanation}") + macd = result.macd + print( + f" MACD: DIF={macd.macd_line} DEA={macd.signal_line} MACD={macd.histogram}" + ) + if verbose and macd.explanation: + print(f" MACD说明: {macd.explanation}") + print(f" RSI(14): {result.rsi.rsi}") + if verbose and result.rsi.explanation: + print(f" RSI说明: {result.rsi.explanation}") + + def error_output(message: str) -> None: """Print error message to stderr.""" print(f"Error: {message}", file=sys.stderr) diff --git a/haoinvest/engine/aggregation.py b/haoinvest/engine/aggregation.py new file mode 100644 index 0000000..c0f80da --- /dev/null +++ b/haoinvest/engine/aggregation.py @@ -0,0 +1,61 @@ +"""Aggregate daily price bars into weekly/monthly bars.""" + +from collections import defaultdict +from datetime import date +from typing import Callable + +from ..models import PriceBar + + +def aggregate_to_weekly(daily_bars: list[PriceBar]) -> list[PriceBar]: + """Aggregate daily bars into weekly bars (ISO week: Monday–Sunday).""" + return _aggregate_bars( + daily_bars, + group_key=lambda d: d.isocalendar()[:2], # (iso_year, iso_week) + ) + + +def aggregate_to_monthly(daily_bars: list[PriceBar]) -> list[PriceBar]: + """Aggregate daily bars into monthly bars (calendar month).""" + return _aggregate_bars( + daily_bars, + group_key=lambda d: (d.year, d.month), + ) + + +def _aggregate_bars( + bars: list[PriceBar], + group_key: Callable[[date], tuple], +) -> list[PriceBar]: + """Group daily bars by key and aggregate OHLCV per group. + + Sorts input by trade_date first so open/close pick the correct first/last bar. + """ + if not bars: + return [] + + sorted_bars = sorted(bars, key=lambda b: b.trade_date) + groups: dict[tuple, list[PriceBar]] = defaultdict(list) + for bar in sorted_bars: + groups[group_key(bar.trade_date)].append(bar) + + result: list[PriceBar] = [] + for group_bars in groups.values(): + highs = [b.high for b in group_bars if b.high is not None] + lows = [b.low for b in group_bars if b.low is not None] + volumes = [b.volume for b in group_bars if b.volume is not None] + + result.append( + PriceBar( + symbol=group_bars[0].symbol, + market_type=group_bars[0].market_type, + trade_date=group_bars[-1].trade_date, + open=group_bars[0].open, + high=max(highs) if highs else None, + low=min(lows) if lows else None, + close=group_bars[-1].close, + volume=sum(volumes) if volumes else None, + ) + ) + + return sorted(result, key=lambda b: b.trade_date) diff --git a/haoinvest/models.py b/haoinvest/models.py index b7ca8ca..914eb7c 100644 --- a/haoinvest/models.py +++ b/haoinvest/models.py @@ -377,11 +377,22 @@ class TechnicalIndicators(BaseModel): macd: MACDResult = Field(default_factory=MACDResult) rsi: RSIResult = Field(default_factory=RSIResult) bollinger: BollingerBands = Field(default_factory=BollingerBands) + timeframe: str = Field(default="daily", description="daily/weekly/monthly") message: Optional[str] = Field( default=None, description="Warning if insufficient data" ) +class MultiTimeframeTechnical(BaseModel): + """Multi-timeframe technical analysis result.""" + + symbol: str + market_type: str + daily: TechnicalIndicators + weekly: Optional[TechnicalIndicators] = None + monthly: Optional[TechnicalIndicators] = None + + class VolumeAnalysis(BaseModel): """Volume anomaly and turnover analysis.""" diff --git a/tests/test_analysis_technical.py b/tests/test_analysis_technical.py index b4b6fb1..ebbc0cf 100644 --- a/tests/test_analysis_technical.py +++ b/tests/test_analysis_technical.py @@ -2,7 +2,7 @@ from datetime import date, timedelta -from haoinvest.analysis.technical import analyze_technical +from haoinvest.analysis.technical import analyze_technical, analyze_technical_multi from haoinvest.db import Database from haoinvest.models import MarketType, PriceBar @@ -124,3 +124,46 @@ def test_math_helpers_not_in_technical_namespace(self): assert not hasattr(t, "_compute_macd") assert not hasattr(t, "_compute_rsi") assert not hasattr(t, "_compute_bollinger") + + +# --------------------------------------------------------------------------- +# Multi-timeframe analysis tests +# --------------------------------------------------------------------------- + + +class TestAnalyzeTechnicalMulti: + def test_all_timeframes_with_sufficient_data(self, db): + """2 years of data → daily/weekly/monthly all populated.""" + _seed_prices(db, days=730, daily_pct=0.002) + result = analyze_technical_multi(db, "TEST", MarketType.A_SHARE) + assert result.symbol == "TEST" + assert result.daily.timeframe == "daily" + assert result.weekly.timeframe == "weekly" + assert result.monthly.timeframe == "monthly" + # Daily should have full indicators + assert result.daily.message is None + assert result.daily.moving_averages.sma_20 is not None + # Weekly should have indicators (730 days ≈ 104 weeks) + assert result.weekly.moving_averages.sma_20 is not None + assert result.weekly.macd.macd_line is not None + # Monthly should have indicators (730 days ≈ 24 months) + assert result.monthly.moving_averages.sma_20 is not None + + def test_short_data_monthly_insufficient(self, db): + """30 days → daily OK, weekly may have limited data, monthly insufficient.""" + _seed_prices(db, days=30) + result = analyze_technical_multi(db, "TEST", MarketType.A_SHARE) + assert result.daily.message is None or "Not enough" not in ( + result.daily.message or "" + ) + # Monthly: 30 days = ~1 month bar, way under 14 + assert result.monthly.message is not None + assert "Not enough" in result.monthly.message + + def test_timeframe_labels_correct(self, db): + """Verify timeframe field is set correctly on each result.""" + _seed_prices(db, days=60) + result = analyze_technical_multi(db, "TEST", MarketType.A_SHARE) + assert result.daily.timeframe == "daily" + assert result.weekly.timeframe == "weekly" + assert result.monthly.timeframe == "monthly" diff --git a/tests/test_engine/test_aggregation.py b/tests/test_engine/test_aggregation.py new file mode 100644 index 0000000..a492779 --- /dev/null +++ b/tests/test_engine/test_aggregation.py @@ -0,0 +1,172 @@ +"""Tests for daily → weekly/monthly price bar aggregation.""" + +from datetime import date + +from haoinvest.engine.aggregation import aggregate_to_monthly, aggregate_to_weekly +from haoinvest.models import MarketType, PriceBar + + +def _bar( + trade_date: str, o: float, h: float, low: float, c: float, v: float +) -> PriceBar: + """Helper to create a PriceBar with defaults.""" + return PriceBar( + symbol="600519", + market_type=MarketType.A_SHARE, + trade_date=date.fromisoformat(trade_date), + open=o, + high=h, + low=low, + close=c, + volume=v, + ) + + +# --- Weekly aggregation --- + + +class TestAggregateToWeekly: + def test_two_weeks(self): + """10 trading days spanning 2 ISO weeks → 2 weekly bars.""" + # Week 1: 2026-03-30 (Mon) to 2026-04-03 (Fri) + # Week 2: 2026-04-06 (Mon) to 2026-04-10 (Fri) — but only 4 days here + bars = [ + _bar("2026-03-30", 100, 105, 98, 102, 1000), + _bar("2026-03-31", 102, 108, 101, 107, 1200), + _bar("2026-04-01", 107, 110, 106, 109, 1100), + _bar("2026-04-02", 109, 112, 108, 111, 900), + _bar("2026-04-03", 111, 115, 110, 113, 1300), + _bar("2026-04-06", 113, 116, 112, 114, 800), + _bar("2026-04-07", 114, 118, 113, 117, 1500), + _bar("2026-04-08", 117, 120, 115, 119, 1000), + _bar("2026-04-09", 119, 121, 118, 120, 1100), + ] + result = aggregate_to_weekly(bars) + assert len(result) == 2 + + w1, w2 = result + # Week 1 + assert w1.trade_date == date(2026, 4, 3) # last trading day + assert w1.open == 100 + assert w1.high == 115 + assert w1.low == 98 + assert w1.close == 113 + assert w1.volume == 5500 + + # Week 2 + assert w2.trade_date == date(2026, 4, 9) + assert w2.open == 113 + assert w2.high == 121 + assert w2.low == 112 + assert w2.close == 120 + assert w2.volume == 4400 + + def test_empty_input(self): + assert aggregate_to_weekly([]) == [] + + def test_single_bar(self): + bars = [_bar("2026-04-01", 100, 105, 95, 102, 500)] + result = aggregate_to_weekly(bars) + assert len(result) == 1 + assert result[0].open == 100 + assert result[0].close == 102 + + def test_unsorted_input(self): + """Bars passed out of order should still aggregate correctly.""" + bars = [ + _bar("2026-04-02", 109, 112, 108, 111, 900), + _bar("2026-04-01", 107, 110, 106, 109, 1100), + ] + result = aggregate_to_weekly(bars) + assert len(result) == 1 + assert result[0].open == 107 # first by date + assert result[0].close == 111 # last by date + + def test_iso_week_year_boundary(self): + """Dec 29, 2025 (Mon) and Jan 2, 2026 (Fri) are in ISO week 2026-W01.""" + bars = [ + _bar("2025-12-29", 50, 55, 48, 52, 100), + _bar("2025-12-30", 52, 56, 51, 54, 200), + _bar("2026-01-02", 54, 58, 53, 57, 300), + ] + result = aggregate_to_weekly(bars) + # All three should be in the same ISO week + assert len(result) == 1 + assert result[0].open == 50 + assert result[0].close == 57 + assert result[0].volume == 600 + + +# --- Monthly aggregation --- + + +class TestAggregateToMonthly: + def test_two_months(self): + """Bars spanning March and April → 2 monthly bars.""" + bars = [ + _bar("2026-03-30", 100, 105, 98, 102, 1000), + _bar("2026-03-31", 102, 108, 101, 107, 1200), + _bar("2026-04-01", 107, 110, 106, 109, 1100), + _bar("2026-04-02", 109, 112, 108, 111, 900), + _bar("2026-04-03", 111, 115, 110, 113, 1300), + ] + result = aggregate_to_monthly(bars) + assert len(result) == 2 + + m1, m2 = result + assert m1.trade_date == date(2026, 3, 31) + assert m1.open == 100 + assert m1.high == 108 + assert m1.low == 98 + assert m1.close == 107 + assert m1.volume == 2200 + + assert m2.trade_date == date(2026, 4, 3) + assert m2.open == 107 + assert m2.close == 113 + assert m2.volume == 3300 + + def test_empty_input(self): + assert aggregate_to_monthly([]) == [] + + +# --- None handling --- + + +class TestNoneHandling: + def test_none_volume(self): + bars = [ + _bar("2026-04-01", 100, 105, 95, 102, 500), + PriceBar( + symbol="600519", + market_type=MarketType.A_SHARE, + trade_date=date(2026, 4, 2), + open=102, + high=108, + low=100, + close=106, + volume=None, + ), + ] + result = aggregate_to_weekly(bars) + assert len(result) == 1 + assert result[0].volume == 500 # only non-None volume summed + + def test_all_none_high_low(self): + bars = [ + PriceBar( + symbol="600519", + market_type=MarketType.A_SHARE, + trade_date=date(2026, 4, 1), + open=100, + high=None, + low=None, + close=102, + volume=None, + ), + ] + result = aggregate_to_weekly(bars) + assert len(result) == 1 + assert result[0].high is None + assert result[0].low is None + assert result[0].volume is None