feat: add analysis-only crypto asset mode

This commit is contained in:
CadeYu
2026-04-18 20:42:11 +08:00
parent f362a160c3
commit e7ec980021
13 changed files with 141 additions and 19 deletions

View File

@@ -505,6 +505,10 @@ def get_user_selections():
)
)
selected_ticker = get_ticker()
asset_type = detect_asset_type(selected_ticker)
console.print(
f"[green]Detected asset type:[/green] {asset_type.value}"
)
# Step 2: Analysis date
default_date = datetime.datetime.now().strftime("%Y-%m-%d")
@@ -523,7 +527,7 @@ def get_user_selections():
"Step 3: Analysts Team", "Select your LLM analyst agents for the analysis"
)
)
selected_analysts = select_analysts()
selected_analysts = select_analysts(asset_type)
console.print(
f"[green]Selected analysts:[/green] {', '.join(analyst.value for analyst in selected_analysts)}"
)
@@ -577,6 +581,7 @@ def get_user_selections():
return {
"ticker": selected_ticker,
"asset_type": asset_type.value,
"analysis_date": analysis_date,
"analysts": selected_analysts,
"research_depth": selected_research_depth,
@@ -989,6 +994,7 @@ def run_analysis():
# Add initial messages
message_buffer.add_message("System", f"Selected ticker: {selections['ticker']}")
message_buffer.add_message("System", f"Detected asset type: {selections['asset_type']}")
message_buffer.add_message(
"System", f"Analysis date: {selections['analysis_date']}"
)
@@ -1011,7 +1017,9 @@ def run_analysis():
# Initialize state and get graph args with callbacks
init_agent_state = graph.propagator.create_initial_state(
selections["ticker"], selections["analysis_date"]
selections["ticker"],
selections["analysis_date"],
asset_type=selections["asset_type"],
)
# Pass callbacks to graph config for tool execution tracking
# (LLM tracking is handled separately via LLM constructor)

View File

@@ -8,3 +8,8 @@ class AnalystType(str, Enum):
SOCIAL = "social"
NEWS = "news"
FUNDAMENTALS = "fundamentals"
class AssetType(str, Enum):
STOCK = "stock"
CRYPTO = "crypto"

View File

@@ -3,7 +3,7 @@ from typing import List, Optional, Tuple, Dict
from rich.console import Console
from cli.models import AnalystType
from cli.models import AnalystType, AssetType
console = Console()
@@ -14,6 +14,8 @@ ANALYST_ORDER = [
("Fundamentals Analyst", AnalystType.FUNDAMENTALS),
]
CRYPTO_SUFFIXES = ("-USD", "-USDT", "-USDC", "-BTC", "-ETH")
def get_ticker() -> str:
"""Prompt the user to enter a ticker symbol."""
@@ -35,6 +37,25 @@ def get_ticker() -> str:
return ticker.strip().upper()
def detect_asset_type(ticker: str) -> AssetType:
normalized_ticker = ticker.strip().upper()
if normalized_ticker.endswith(CRYPTO_SUFFIXES):
return AssetType.CRYPTO
return AssetType.STOCK
def filter_analysts_for_asset_type(
analysts: List[AnalystType], asset_type: AssetType
) -> List[AnalystType]:
if asset_type != AssetType.CRYPTO:
return analysts
return [
analyst
for analyst in analysts
if analyst != AnalystType.FUNDAMENTALS
]
def get_analysis_date() -> str:
"""Prompt the user to enter a date in YYYY-MM-DD format."""
import re
@@ -68,12 +89,18 @@ def get_analysis_date() -> str:
return date.strip()
def select_analysts() -> List[AnalystType]:
def select_analysts(asset_type: AssetType = AssetType.STOCK) -> List[AnalystType]:
"""Select analysts using an interactive checkbox."""
available_analysts = filter_analysts_for_asset_type(
[value for _, value in ANALYST_ORDER],
asset_type,
)
choices = questionary.checkbox(
"Select Your [Analysts Team]:",
choices=[
questionary.Choice(display, value=value) for display, value in ANALYST_ORDER
questionary.Choice(display, value=value)
for display, value in ANALYST_ORDER
if value in available_analysts
],
instruction="\n- Press Space to select/unselect analysts\n- Press 'a' to select/unselect all\n- Press Enter when done",
validate=lambda x: len(x) > 0 or "You must select at least one analyst.",