Skip to content

A
ai-stock-analysis
Commit 63580797 authored by xeron56's avatar xeron56
Browse files
Options

feat: Introduce AI-driven analysis and reporting for stock research 

- Added AITraderReport and AIResearch models to encapsulate AI-generated trading reports and research details.
- Enhanced StockAnalysis model to include optional AI research data.
- Implemented AIStockAnalysisGenerator for generating AI-enhanced stock analysis.
- Created grounded prompts for AI stock research in a new stock_research.py file.
- Developed a lightweight dashboard CLI for fetching and displaying stock analysis without starting the multi-agent graph.
- Updated AnalysisService to integrate AI analysis into the stock analysis workflow.
- Added tests for AI dashboard analysis and ensured proper functionality of new features.
parent 6132f5c0
Hide whitespace changes
Inline Side-by-side
Showing with 1629 additions and 43 deletions
README.md
View file @ 63580797
......@@ -336,6 +336,24 @@ Environment overrides are applied to `DEFAULT_CONFIG` before this code runs.
## REST API and local dashboard
To fetch the yearly DSE evidence, run one grounded AI synthesis for the score,
valuation weighting, full research, and trader view, then open the UI without
running the long multi-agent graph, use:
```bash
dohasecuritiesstockai-dashboard GP --date 2026-08-10
```
The command sends only the collected, date-bounded evidence to the configured
deep-thinking model and requires that provider's API key. Numeric valuation
methods remain calculation-backed; the AI assigns reliability weights, writes
the research, and produces the trader view. The first UI launch may still need
time to install/build the Angular frontend. Use `--no-open-ui` to generate and
save the AI payload without starting the server, or `--no-ai` for the original
calculation-only fallback.
The equivalent source-tree command is
`python -m dohasecuritiesstockai.dashboard_cli GP --date 2026-08-10`.
Start the read-only analysis API:
```bash
......
analysis-ui/src/app/app.component.html
View file @ 63580797
......@@ -67,6 +67,9 @@
<h1>{{ report.company_name }}</h1>
<div class="company-tags">
<span class="symbol-tag">{{ report.symbol }}</span>
@if (report.ai_research) {
<span class="ai-tag">AI-grounded</span>
}
<span>{{ report.sector }}</span>
<span>{{ language() === 'bn' ? 'সর্বশেষ দাম' : 'Latest price' }}: {{ formatTaka(report.market.latest_price) }}</span>
@if (report.market.change_percent !== null) {
......@@ -84,7 +87,7 @@
<section class="score-overview">
<article class="score-card">
<p class="section-label">{{ language() === 'bn' ? 'মৌলিক স্কোর' : 'Fundamental score' }}</p>
<p class="section-label">{{ report.ai_research ? (language() === 'bn' ? 'এআই মৌলিক স্কোর' : 'AI fundamental score') : (language() === 'bn' ? 'মৌলিক স্কোর' : 'Fundamental score') }}</p>
<div class="score-chart">
<canvas #scoreCanvas aria-label="Fundamental score chart"></canvas>
<div class="score-center">
......@@ -103,6 +106,13 @@
<li>{{ copy(takeaway) }}</li>
}
</ul>
@if (report.ai_research; as ai) {
<p class="ai-rationale">{{ copy(ai.score_rationale) }}</p>
<small class="ai-provenance">
{{ language() === 'bn' ? 'এআই মডেল' : 'AI model' }}: {{ ai.provider }} / {{ ai.model }} ·
{{ language() === 'bn' ? 'আস্থা' : 'confidence' }}: {{ ai.score_confidence }}
</small>
}
</article>
</section>
......@@ -146,6 +156,13 @@
}
<p class="valuation-summary">{{ copy(report.valuation.summary) }}</p>
@if (report.ai_research) {
<p class="valuation-formula">
{{ language() === 'bn'
? 'এআই প্রতিটি হিসাব-পদ্ধতির নির্ভরযোগ্যতার ওজন নির্ধারণ করেছে; নিচের সংখ্যাগুলো যাচাইযোগ্য সূত্রে গণনা করা। ন্যায্য সীমা আনুমানিক মূল্যের ৮০%–১২০%।'
: 'AI assigned reliability weights to each method; the figures below remain calculation-backed. The fair range is 80%–120% of the rough estimate.' }}
</p>
}
<dl class="valuation-methods">
@for (method of report.valuation.methods; track method.key) {
<div [class.unavailable]="!method.available">
......@@ -203,6 +220,59 @@
}
</div>
@if (report.ai_research; as ai) {
<section class="ai-trader-report">
<header>
<div>
<p class="section-label">{{ language() === 'bn' ? 'এআই ট্রেডার ভিউ' : 'AI trader view' }}</p>
<h2>{{ copy(ai.trader_report.action) }}</h2>
</div>
<span [class]="'trader-rating ' + ai.trader_report.rating.toLowerCase()">
{{ ai.trader_report.rating }}
</span>
</header>
<p class="trader-summary">{{ copy(ai.trader_report.executive_summary) }}</p>
<div class="trader-grid">
<article>
<h3>{{ language() === 'bn' ? 'বিনিয়োগ যুক্তি' : 'Investment thesis' }}</h3>
<p>{{ copy(ai.trader_report.investment_thesis) }}</p>
</article>
<article>
<h3>{{ language() === 'bn' ? 'এন্ট্রি কৌশল' : 'Entry strategy' }}</h3>
<p>{{ copy(ai.trader_report.entry_strategy) }}</p>
</article>
<article>
<h3>{{ language() === 'bn' ? 'ঝুঁকি নিয়ন্ত্রণ' : 'Risk controls' }}</h3>
<p>{{ copy(ai.trader_report.risk_controls) }}</p>
</article>
<article>
<h3>{{ language() === 'bn' ? 'সময়সীমা' : 'Time horizon' }}</h3>
<p>{{ copy(ai.trader_report.time_horizon) }}</p>
</article>
</div>
<div class="trader-lists">
<div>
<h3>{{ language() === 'bn' ? 'সম্ভাব্য অনুঘটক' : 'Catalysts' }}</h3>
<ul>
@for (item of ai.trader_report.catalysts; track item.en) {
<li>{{ copy(item) }}</li>
}
</ul>
</div>
<div>
<h3>{{ language() === 'bn' ? 'যে অবস্থায় মত বদলাবে' : 'Invalidation conditions' }}</h3>
<ul>
@for (item of ai.trader_report.invalidation_conditions; track item.en) {
<li>{{ copy(item) }}</li>
}
</ul>
</div>
</div>
<p class="data-quality"><strong>{{ language() === 'bn' ? 'তথ্যের মান:' : 'Data quality:' }}</strong> {{ copy(ai.data_quality) }}</p>
</section>
}
@if (hasAgentEvidence(report)) {
<section class="agent-evidence">
<h2>{{ language() === 'bn' ? 'মূল এজেন্ট প্রমাণ' : 'Original agent evidence' }}</h2>
<p>{{ language() === 'bn' ? 'নিচের অংশগুলো মূল মাল্টি-এজেন্ট রান থেকে অপরিবর্তিত রাখা হয়েছে।' : 'These sections are retained unchanged from the multi-agent run.' }}</p>
......@@ -220,6 +290,7 @@
}
<details><summary>{{ language() === 'bn' ? 'পূর্ণ স্টেট লগ JSON' : 'Full state log JSON' }}</summary><pre class="json-state">{{ rawState(report) }}</pre></details>
</section>
}
</section>
}
......
analysis-ui/src/app/app.component.scss
View file @ 63580797
......@@ -186,6 +186,12 @@
background: #f0f6f9;
}
.company-tags .ai-tag {
border-color: #9bcab8;
color: var(--green);
background: #eef8f3;
}
.company-tags .negative-change {
color: var(--red);
}
......@@ -275,6 +281,22 @@
color: var(--blue);
}
.ai-rationale {
max-width: 62rem;
margin: 1.25rem 0 0;
padding: 1rem 1.15rem;
border-left: 0.24rem solid var(--blue);
color: var(--muted);
background: #f3f7f9;
line-height: 1.55;
}
.ai-provenance {
display: block;
margin-top: 0.65rem;
color: var(--muted);
}
.in-depth-card,
.value-card,
.factor-card,
......@@ -388,6 +410,16 @@
line-height: 1.55;
}
.valuation-formula {
max-width: 72rem;
padding: 0.8rem 1rem;
border-radius: 0.8rem;
color: #456759;
background: #eef7f2;
font-size: 0.92rem;
line-height: 1.5;
}
.valuation-methods {
display: grid;
gap: 0.55rem;
......@@ -550,6 +582,76 @@
margin-top: 2.5rem;
}
.ai-trader-report {
margin-top: 2.5rem;
padding: clamp(1.3rem, 2.5vw, 2rem);
border: 1px solid #b9d3c7;
border-radius: 1.25rem;
background: #f2f8f5;
}
.ai-trader-report > header {
display: flex;
align-items: flex-start;
justify-content: space-between;
gap: 1rem;
}
.ai-trader-report h2,
.ai-trader-report h3 {
margin-top: 0;
}
.trader-rating {
padding: 0.45rem 1rem;
border-radius: 2rem;
color: #fff;
background: var(--green);
font-weight: 750;
}
.trader-rating.hold {
background: var(--copper);
}
.trader-rating.underweight,
.trader-rating.sell {
background: var(--red);
}
.trader-summary {
max-width: 72rem;
color: var(--ink);
font-size: 1.1rem;
line-height: 1.6;
}
.trader-grid,
.trader-lists {
display: grid;
grid-template-columns: repeat(2, minmax(0, 1fr));
gap: 1rem;
margin-top: 1rem;
}
.trader-grid article,
.trader-lists > div {
padding: 1.1rem;
border-radius: 0.9rem;
background: rgb(255 255 255 / 75%);
}
.trader-grid p,
.trader-lists li,
.data-quality {
color: var(--muted);
line-height: 1.55;
}
.data-quality {
margin: 1rem 0 0;
}
.agent-evidence > p {
color: var(--muted);
}
......@@ -678,7 +780,9 @@
}
.score-overview,
.report-section-grid {
.report-section-grid,
.trader-grid,
.trader-lists {
grid-template-columns: 1fr;
}
......
analysis-ui/src/app/app.component.ts
View file @ 63580797
......@@ -155,6 +155,18 @@ export class AppComponent implements OnInit, AfterViewInit, OnDestroy {
return JSON.stringify(report.agent_reports.raw_state, null, 2);
}
hasAgentEvidence(report: StockAnalysis): boolean {
const evidence = report.agent_reports;
return Boolean(
evidence.market_report ||
evidence.news_report ||
evidence.fundamentals_report ||
evidence.investment_plan ||
evidence.final_trade_decision ||
Object.keys(evidence.raw_state).length,
);
}
private loadStocks(): void {
this.loadingStocks.set(true);
this.subscriptions.push(
......
analysis-ui/src/app/stock-analysis.model.ts
View file @ 63580797
......@@ -76,6 +76,31 @@ export interface AgentReports {
raw_state: Record<string, unknown>;
}
export interface AITraderReport {
rating: 'Buy' | 'Overweight' | 'Hold' | 'Underweight' | 'Sell';
action: BilingualText;
confidence: 'low' | 'medium' | 'high';
executive_summary: BilingualText;
investment_thesis: BilingualText;
entry_strategy: BilingualText;
risk_controls: BilingualText;
catalysts: BilingualText[];
invalidation_conditions: BilingualText[];
time_horizon: BilingualText;
}
export interface AIResearch {
provider: string;
model: string;
mode: 'ai_fundamental' | 'multi_agent_synthesis';
generated_at: string;
score_confidence: 'low' | 'medium' | 'high';
score_rationale: BilingualText;
data_quality: BilingualText;
valuation_method_weights: Record<string, number>;
trader_report: AITraderReport;
}
export interface StockAnalysis {
schema_version: '1.0';
analysis_id: string;
......@@ -95,6 +120,7 @@ export interface StockAnalysis {
factors: FactorCard[];
report_sections: ReportSection[];
agent_reports: AgentReports;
ai_research?: AIResearch | null;
disclaimer: BilingualText;
}
......
dohasecuritiesstockai/api/ai_analysis.py 0 → 100644
View file @ 63580797
"""Grounded AI synthesis for the score dashboard and full analysis page."""
from __future__ import annotations
import json
import re
from datetime import datetime, timezone
from typing import Any, Literal
from pydantic import BaseModel, Field, model_validator
from dohasecuritiesstockai.default_config import DEFAULT_CONFIG
from dohasecuritiesstockai.llm_clients import create_llm_client
from .models import (
AIResearch,
AITraderReport,
BilingualText,
EvidenceSource,
ReportSection,
ScoreMetric,
StockAnalysis,
)
from .prompts.stock_research import (
AI_STOCK_RESEARCH_SYSTEM_PROMPT,
AI_STOCK_RESEARCH_TASK_TEMPLATE,
)
_SECTION_KEYS = (
"company",
"business_model",
"profitability",
"financial_safety",
"valuation",
"dividends",
"moat",
"bull_case",
"risks",
"suitability",
)
_FACTOR_WEIGHTS = {
"profitability": 2.5,
"financial_health": 2.5,
"business_quality": 2.0,
"valuation": 1.5,
"dividend": 1.5,
}
_METHOD_KEYS = ("historical_pe", "peer_pe", "historical_pb", "dividend_yield")
_NARRATIVE_PATTERNS = {
"overall /100 score": re.compile(
r"(?:\d+(?:\.\d+)?\s*/\s*100|[০-৯]+\s*/\s*১০০)", re.IGNORECASE
),
"not-yet-calculated aggregate valuation": re.compile(
r"(?:rough estimate|fair range|weighted (?:rough )?estimate|"
r"আনুমানিক মূল্য|ন্যায্য পরিসর)",
re.IGNORECASE,
),
"unsupported scaling for raw gateway financials": re.compile(
r"(?:paid-up capital|reserve(?:s| and surplus)?|net profit|annual profit|"
r"cash balance|cash & cash equivalents|total equity|total assets|liabilit(?:y|ies))"
r"[^.\n]{0,80}\b(?:crore|million|billion|lakh)\b",
re.IGNORECASE,
),
}
class AIText(BaseModel):
en: str = Field(min_length=1)
bn: str = Field(min_length=1)
def api_text(self) -> BilingualText:
return BilingualText(en=self.en.strip(), bn=self.bn.strip())
class AIFactorAssessment(BaseModel):
score: float = Field(ge=0, le=10)
rationale: AIText
class AIFactorAssessments(BaseModel):
profitability: AIFactorAssessment
financial_health: AIFactorAssessment
business_quality: AIFactorAssessment
valuation: AIFactorAssessment
dividend: AIFactorAssessment
class AIValuationWeights(BaseModel):
historical_pe: float = Field(ge=0, le=1)
peer_pe: float = Field(ge=0, le=1)
historical_pb: float = Field(ge=0, le=1)
dividend_yield: float = Field(ge=0, le=1)
class AIValuationAssessment(BaseModel):
weights: AIValuationWeights
confidence: Literal["low", "medium", "high"]
summary: AIText
class AIReportSectionOutput(BaseModel):
key: Literal[
"company",
"business_model",
"profitability",
"financial_safety",
"valuation",
"dividends",
"moat",
"bull_case",
"risks",
"suitability",
]
title: AIText
summary: AIText
bullets: list[AIText] = Field(default_factory=list, max_length=6)
class AITraderOutput(BaseModel):
rating: Literal["Buy", "Overweight", "Hold", "Underweight", "Sell"]
action: AIText
confidence: Literal["low", "medium", "high"]
executive_summary: AIText
investment_thesis: AIText
entry_strategy: AIText
risk_controls: AIText
catalysts: list[AIText] = Field(min_length=2, max_length=5)
invalidation_conditions: list[AIText] = Field(min_length=2, max_length=5)
time_horizon: AIText
class AIStockResearchOutput(BaseModel):
factors: AIFactorAssessments
score_confidence: Literal["low", "medium", "high"]
score_rationale: AIText
headline: AIText
takeaways: list[AIText] = Field(min_length=3, max_length=5)
in_depth_title: AIText
in_depth_summary: AIText
valuation: AIValuationAssessment
sections: list[AIReportSectionOutput] = Field(min_length=10, max_length=10)
trader: AITraderOutput
data_quality: AIText
@model_validator(mode="after")
def require_every_section(self):
keys = [section.key for section in self.sections]
if len(set(keys)) != len(keys) or set(keys) != set(_SECTION_KEYS):
raise ValueError(f"sections must contain exactly: {', '.join(_SECTION_KEYS)}")
return self
def _provider_kwargs(config: dict[str, Any]) -> dict[str, Any]:
provider = str(config.get("llm_provider", "")).lower()
kwargs: dict[str, Any] = {}
if provider == "google" and config.get("google_thinking_level"):
kwargs["thinking_level"] = config["google_thinking_level"]
elif provider == "openai" and config.get("openai_reasoning_effort"):
kwargs["reasoning_effort"] = config["openai_reasoning_effort"]
elif provider == "anthropic" and config.get("anthropic_effort"):
kwargs["effort"] = config["anthropic_effort"]
temperature = config.get("temperature")
kwargs["temperature"] = 0.1 if temperature in (None, "") else float(temperature)
retries = config.get("llm_max_retries")
if retries not in (None, ""):
kwargs["max_retries"] = max(0, int(retries))
return kwargs
def _clip(value: Any, limit: int = 16_000) -> str:
text = str(value or "")
return text if len(text) <= limit else text[:limit] + "\n[truncated]"
def _agent_evidence(state: dict[str, Any] | None) -> dict[str, str]:
if not state:
return {}
keys = (
"market_report",
"news_report",
"fundamentals_report",
"investment_plan",
"trader_investment_decision",
"trader_investment_plan",
"final_trade_decision",
)
return {key: _clip(state.get(key)) for key in keys if state.get(key)}
def _extract_json(content: str) -> dict[str, Any]:
stripped = content.strip()
fenced = re.search(r"```(?:json)?\s*(\{.*\})\s*```", stripped, re.DOTALL)
if fenced:
stripped = fenced.group(1)
else:
start, end = stripped.find("{"), stripped.rfind("}")
if start >= 0 and end > start:
stripped = stripped[start : end + 1]
payload = json.loads(stripped)
if not isinstance(payload, dict):
raise ValueError("AI research response was not a JSON object")
return payload
def _narrative_violations(result: AIStockResearchOutput) -> list[str]:
"""Find prose that claims values calculated only after the model response."""
violations: list[str] = []
def walk(value: Any, path: str = "output") -> None:
if isinstance(value, dict):
for key, item in value.items():
walk(item, f"{path}.{key}")
elif isinstance(value, list):
for index, item in enumerate(value):
walk(item, f"{path}[{index}]")
elif isinstance(value, str):
for label, pattern in _NARRATIVE_PATTERNS.items():
if pattern.search(value):
violations.append(f"{path}: {label}")
walk(result.model_dump(mode="json"))
return violations
class AIStockAnalysisGenerator:
"""Generate and validate one structured AI research synthesis."""
def __init__(
self,
config: dict[str, Any] | None = None,
*,
llm: Any | None = None,
) -> None:
self.config = config or DEFAULT_CONFIG
self.provider = str(self.config["llm_provider"])
self.model = str(self.config["deep_think_llm"])
if llm is None:
client = create_llm_client(
provider=self.provider,
model=self.model,
base_url=self.config.get("backend_url"),
**_provider_kwargs(self.config),
)
llm = client.get_llm()
self.llm = llm
def _invoke(self, messages: list[dict[str, str]]) -> AIStockResearchOutput:
try:
structured = self.llm.with_structured_output(AIStockResearchOutput)
result = structured.invoke(messages)
if result is not None:
parsed = AIStockResearchOutput.model_validate(result)
violations = _narrative_violations(parsed)
if not violations:
return parsed
correction = {
"role": "user",
"content": (
"Your prior structured response violated these grounding rules:\n- "
+ "\n- ".join(violations)
+ "\nReturn a corrected full structured object. Preserve supported evidence, "
"but remove every prohibited aggregate claim or unsupported unit.\n\n"
"PRIOR RESPONSE:\n"
+ parsed.model_dump_json()
),
}
corrected = AIStockResearchOutput.model_validate(
structured.invoke([*messages, correction])
)
remaining = _narrative_violations(corrected)
if remaining:
raise ValueError(
"AI research narrative remained ungrounded after correction: "
+ "; ".join(remaining)
)
return corrected
except (NotImplementedError, AttributeError):
pass
json_messages = [
*messages,
{
"role": "user",
"content": (
"Structured-output binding is unavailable. Return only one JSON object "
"that validates against the requested schema; no markdown fences or prose."
),
},
]
response = self.llm.invoke(json_messages)
parsed = AIStockResearchOutput.model_validate(_extract_json(response.content))
violations = _narrative_violations(parsed)
if violations:
raise ValueError(
"AI research narrative violated grounding rules: " + "; ".join(violations)
)
return parsed
def enhance(
self,
analysis: StockAnalysis,
evidence: dict[str, Any],
agent_state: dict[str, Any] | None = None,
) -> StockAnalysis:
mode = "multi_agent_synthesis" if agent_state else "ai_fundamental"
prompt_evidence = {
**evidence,
"multi_agent_reports": _agent_evidence(agent_state),
}
task = AI_STOCK_RESEARCH_TASK_TEMPLATE.format(
symbol=analysis.symbol,
analysis_date=analysis.analysis_date.isoformat(),
mode=mode,
evidence_json=json.dumps(prompt_evidence, ensure_ascii=False, default=str),
)
result = self._invoke(
[
{"role": "system", "content": AI_STOCK_RESEARCH_SYSTEM_PROMPT},
{"role": "user", "content": task},
]
)
return self._apply(analysis, result, mode)
def _apply(
self,
analysis: StockAnalysis,
result: AIStockResearchOutput,
mode: Literal["ai_fundamental", "multi_agent_synthesis"],
) -> StockAnalysis:
factor_outputs = result.factors.model_dump()
factor_scores = {
key: float(value["score"])
for key, value in factor_outputs.items()
}
score = round(
sum(factor_scores[key] * weight for key, weight in _FACTOR_WEIGHTS.items())
)
score = max(0, min(100, score))
if score >= 75:
score_label = BilingualText(en="Very good", bn="খুব ভালো")
elif score >= 60:
score_label = BilingualText(en="Good", bn="ভালো")
elif score >= 45:
score_label = BilingualText(en="Mixed", bn="মিশ্র")
else:
score_label = BilingualText(en="Weak", bn="দুর্বল")
factors = []
for factor in analysis.factors:
assessment = getattr(result.factors, factor.key)
ai_metric = ScoreMetric(
key="ai_factor_assessment",
label=BilingualText(en="AI factor judgment", bn="এআই ফ্যাক্টর মূল্যায়ন"),
display_value=f"{assessment.score:.1f}/10",
score=assessment.score,
)
factors.append(
factor.model_copy(
update={
"status": (
"positive"
if assessment.score >= 7
else "caution"
if assessment.score >= 4
else "negative"
),
"explanation": assessment.rationale.api_text(),
"metrics": [ai_metric, *factor.metrics],
}
)
)
valuation, method_weights = self._weighted_valuation(analysis, result)
ordered_sections = {section.key: section for section in result.sections}
sections = [
ReportSection(
key=key,
title=ordered_sections[key].title.api_text(),
summary=ordered_sections[key].summary.api_text(),
bullets=[bullet.api_text() for bullet in ordered_sections[key].bullets],
)
for key in _SECTION_KEYS
]
trader = result.trader
ai_research = AIResearch(
provider=self.provider,
model=self.model,
mode=mode,
generated_at=datetime.now(timezone.utc),
score_confidence=result.score_confidence,
score_rationale=result.score_rationale.api_text(),
data_quality=result.data_quality.api_text(),
valuation_method_weights=method_weights,
trader_report=AITraderReport(
rating=trader.rating,
action=trader.action.api_text(),
confidence=trader.confidence,
executive_summary=trader.executive_summary.api_text(),
investment_thesis=trader.investment_thesis.api_text(),
entry_strategy=trader.entry_strategy.api_text(),
risk_controls=trader.risk_controls.api_text(),
catalysts=[item.api_text() for item in trader.catalysts],
invalidation_conditions=[
item.api_text() for item in trader.invalidation_conditions
],
time_horizon=trader.time_horizon.api_text(),
),
)
ai_source = EvidenceSource(
name=f"{self.provider}:{self.model}",
source_type="ai_analysis",
detail=(
"Structured AI synthesis of the supplied date-bounded DSE evidence; "
"numeric valuation anchors remain calculation-backed."
),
)
return analysis.model_copy(
update={
"fundamental_score": score,
"score_label": score_label,
"headline": result.headline.api_text(),
"takeaways": [item.api_text() for item in result.takeaways],
"in_depth_title": result.in_depth_title.api_text(),
"in_depth_snippet": result.in_depth_summary.api_text(),
"valuation": valuation,
"factors": factors,
"report_sections": sections,
"ai_research": ai_research,
"sources": [*analysis.sources, ai_source],
}
)
@staticmethod
def _weighted_valuation(
analysis: StockAnalysis,
result: AIStockResearchOutput,
) -> tuple[Any, dict[str, float]]:
methods = {method.key: method for method in analysis.valuation.methods}
requested = result.valuation.weights.model_dump()
usable = {
key: methods[key].value
for key in _METHOD_KEYS
if key in methods and methods[key].available and methods[key].value is not None
}
raw_weights = {key: requested[key] if key in usable else 0.0 for key in _METHOD_KEYS}
total = sum(raw_weights.values())
if usable and total <= 0:
raw_weights = {key: (1.0 if key in usable else 0.0) for key in _METHOD_KEYS}
total = float(len(usable))
normalized = {
key: round(raw_weights[key] / total, 4) if total else 0.0
for key in _METHOD_KEYS
}
estimate = (
sum(float(usable[key]) * normalized[key] for key in usable)
if usable
else None
)
estimate = round(estimate, 1) if estimate else None
fair_low = round(estimate * 0.8, 1) if estimate else None
fair_high = round(estimate * 1.2, 1) if estimate else None
price = analysis.market.latest_price
if estimate is None:
verdict = "insufficient_data"
label = BilingualText(en="Not enough data", bn="পর্যাপ্ত তথ্য নেই")
elif price <= estimate * 0.85:
verdict = "looks_cheap"
label = BilingualText(en="Looks cheap", bn="সস্তা মনে হচ্ছে")
elif price >= estimate * 1.15:
verdict = "looks_expensive"
label = BilingualText(en="Looks expensive", bn="দামি মনে হচ্ছে")
else:
verdict = "fair"
label = BilingualText(en="Fair price", bn="ন্যায্য দাম")
valuation = analysis.valuation.model_copy(
update={
"verdict": verdict,
"verdict_label": label,
"rough_estimate": estimate,
"fair_range_low": fair_low,
"fair_range_high": fair_high,
"confidence": result.valuation.confidence,
"summary": result.valuation.summary.api_text(),
}
)
return valuation, normalized
dohasecuritiesstockai/api/analysis.py
View file @ 63580797
......@@ -49,6 +49,16 @@ def _number(value: Any) -> float | None:
return number if math.isfinite(number) else None
def _market_price(row: dict[str, Any]) -> float | None:
"""Return the best positive quote, including the market-closed fallback."""
for key in ("ltp", "close", "ycp"):
value = _number(row.get(key))
if value is not None and value > 0:
return value
return None
def _clamp(value: float, low: float = 0, high: float = 10) -> float:
return round(max(low, min(high, value)), 1)
......@@ -83,6 +93,61 @@ def _annual_rows(rows: Any, cutoff_year: int) -> list[dict[str, Any]]:
return sorted(annual, key=_year)
def _rows_through(rows: Any, cutoff: date) -> list[dict[str, Any]]:
"""Return disclosure rows that were available on or before ``cutoff``."""
if not isinstance(rows, list):
return []
accepted: list[dict[str, Any]] = []
for row in rows:
if not isinstance(row, dict):
continue
raw_date = row.get("date") or row.get("record_date") or row.get("published_at")
if raw_date:
try:
parsed = datetime.fromisoformat(str(raw_date).replace("Z", "+00:00")).date()
except ValueError:
parsed = None
if parsed is not None and parsed > cutoff:
continue
row_year = _year(row)
if row_year and row_year > cutoff.year:
continue
accepted.append(row)
return accepted
def _balance_history_through(balance: Any, cutoff_year: int) -> list[dict[str, Any]]:
"""Turn the gateway's columnar balance sheet into year-keyed evidence."""
if not isinstance(balance, dict):
return []
columns = balance.get("columns")
histories = balance.get("year_wise_data")
if not isinstance(columns, list) or not isinstance(histories, list):
return []
results: list[dict[str, Any]] = []
for item in histories:
if not isinstance(item, dict):
continue
for raw_year, values in item.items():
try:
year = int(str(raw_year)[:4])
except (TypeError, ValueError):
continue
if 0 < year <= cutoff_year and isinstance(values, list):
results.append(
{
"year": year,
"values": {
str(column): value
for column, value in zip(columns, values, strict=False)
},
}
)
return sorted(results, key=lambda item: item["year"])[-8:]
def _metric(
key: str,
en: str,
......@@ -130,7 +195,7 @@ def list_dse_stocks(client: DSEClient | None = None) -> list[StockOption]:
symbol=symbol,
name=str(row.get("securityName") or symbol).strip(),
sector=str(row.get("sector") or "").strip(),
latest_price=_number(row.get("ltp")),
latest_price=_market_price(row),
change_percent=_number(row.get("changePercentage")),
)
)
......@@ -153,6 +218,7 @@ class StockAnalysisBuilder:
def __init__(self, client: DSEClient | None = None) -> None:
self.client = client or DSEClient()
self.last_evidence: dict[str, Any] = {}
def _fundamental(self, path: str, symbol: str) -> Any:
return _unwrap(
......@@ -244,7 +310,7 @@ class StockAnalysisBuilder:
continue
if peer == symbol:
continue
price = _number(row.get("ltp"))
price = _market_price(row)
if not price or price <= 0:
continue
try:
......@@ -276,7 +342,8 @@ class StockAnalysisBuilder:
self.client.get("analytics", f"/api/balance_sheet/balance-sheet/{symbol}")
)
company = data["company"] if isinstance(data["company"], dict) else {}
price = _number(quote.get("ltp")) or _number(quote.get("close"))
history = self._historical_snapshot(symbol, analysis_date)
price = history.get("latest_price") or _market_price(quote)
if not price or price <= 0:
raise LookupError(f"No usable DSE price was returned for {symbol}")
......@@ -325,6 +392,7 @@ class StockAnalysisBuilder:
str(quote.get("sector") or company.get("sector") or ""),
universe,
cutoff_year,
allow_live_peers=analysis_date >= date.today(),
)
dividend_factor = self._dividend_factor(dividends, price)
factors = [
......@@ -344,8 +412,12 @@ class StockAnalysisBuilder:
)
score = int(max(0, min(100, score)))
momentum = self._momentum(symbol, analysis_date)
momentum = {
"return_30d": history.get("return_30d"),
"quiet": bool(history.get("quiet")),
}
score_label, headline = self._headline(score, momentum)
state = agent_state or {}
report_sections = self._sections(
score,
profit_factor,
......@@ -369,12 +441,17 @@ class StockAnalysisBuilder:
f"{company_name} combines {health_factor.title.en.lower()} with {profit_factor.title.en.lower()} — valuation and execution still matter.",
f"{company_name}-এর {health_factor.title.bn} এবং {profit_factor.title.bn}—তবে মূল্যায়ন ও বাস্তবায়ন গুরুত্বপূর্ণ।",
)
in_depth_snippet = text(
"This report brings together DSE price history, company disclosures, dividends, ownership and the complete multi-agent debate. Open the full analysis to see the evidence, risks and final trading view.",
"এই প্রতিবেদনে ডিএসই মূল্য ইতিহাস, কোম্পানি প্রকাশনা, লভ্যাংশ, মালিকানা এবং সম্পূর্ণ মাল্টি-এজেন্ট বিতর্ক একত্র করা হয়েছে। প্রমাণ, ঝুঁকি ও চূড়ান্ত ট্রেডিং মত দেখতে পূর্ণ বিশ্লেষণ খুলুন।",
in_depth_snippet = (
text(
"This report brings together DSE price history, company disclosures, dividends, ownership and the complete multi-agent debate. Open the full analysis to see the evidence, risks and final trading view.",
"এই প্রতিবেদনে ডিএসই মূল্য ইতিহাস, কোম্পানি প্রকাশনা, লভ্যাংশ, মালিকানা এবং সম্পূর্ণ মাল্টি-এজেন্ট বিতর্ক একত্র করা হয়েছে। প্রমাণ, ঝুঁকি ও চূড়ান্ত ট্রেডিং মত দেখতে পূর্ণ বিশ্লেষণ খুলুন।",
)
if state
else text(
"This lightweight report uses only read-only DSE price history, company disclosures, dividends and ownership data. No multi-agent or LLM analysis was run.",
"এই হালকা প্রতিবেদনে শুধু পঠনযোগ্য ডিএসই মূল্য ইতিহাস, কোম্পানি প্রকাশনা, লভ্যাংশ ও মালিকানার তথ্য ব্যবহার করা হয়েছে। কোনো মাল্টি-এজেন্ট বা এলএলএম বিশ্লেষণ চালানো হয়নি।",
)
)
state = agent_state or {}
agent_reports = AgentReports(
market_report=str(state.get("market_report") or ""),
news_report=str(state.get("news_report") or ""),
......@@ -387,8 +464,21 @@ class StockAnalysisBuilder:
final_trade_decision=str(state.get("final_trade_decision") or ""),
raw_state=state,
)
low_52, high_52 = self._parse_range(company.get("fifty_two_weeks_moving_range"))
return StockAnalysis(
low_52 = history.get("fifty_two_week_low")
high_52 = history.get("fifty_two_week_high")
if low_52 is None or high_52 is None:
low_52, high_52 = self._parse_range(
company.get("fifty_two_weeks_moving_range")
)
previous_close = history.get("previous_close") or _number(quote.get("ycp"))
price_change = history.get("change")
change_percent = history.get("change_percent")
if price_change is None:
price_change = _number(quote.get("change"))
if change_percent is None:
change_percent = _number(quote.get("changePercentage"))
market_as_of = history.get("as_of") or datetime.now(timezone.utc)
analysis = StockAnalysis(
analysis_id=f"{symbol}-{analysis_date}",
symbol=symbol,
company_name=company_name,
......@@ -397,12 +487,12 @@ class StockAnalysisBuilder:
generated_at=datetime.now(timezone.utc),
market=MarketSnapshot(
latest_price=price,
change=_number(quote.get("change")),
change_percent=_number(quote.get("changePercentage")),
previous_close=_number(quote.get("ycp")),
change=price_change,
change_percent=change_percent,
previous_close=previous_close,
fifty_two_week_low=low_52,
fifty_two_week_high=high_52,
as_of=datetime.now(timezone.utc),
as_of=market_as_of,
),
fundamental_score=score,
score_label=score_label,
......@@ -418,12 +508,18 @@ class StockAnalysisBuilder:
EvidenceSource(
name="Doha Securities DSE gateway",
source_type="dse_api",
detail="Live quote, candles, company fundamentals, balance sheet, ownership and dividends (read-only).",
detail="Date-bounded candles, company fundamentals, balance sheet, ownership and dividends (read-only).",
),
EvidenceSource(
name="TradingAgents full state",
source_type="agent_state",
detail="Market, news and fundamentals reports, researcher debate, risk debate and final decision.",
*(
[
EvidenceSource(
name="TradingAgents full state",
source_type="agent_state",
detail="Market, news and fundamentals reports, researcher debate, risk debate and final decision.",
)
]
if state
else []
),
EvidenceSource(
name="Transparent presentation calculations",
......@@ -436,6 +532,34 @@ class StockAnalysisBuilder:
"শুধু শিক্ষামূলক বিশ্লেষণ। ন্যায্য মূল্য উপলভ্য তথ্যের আনুমানিক হিসাব; এটি মূল্য লক্ষ্য বা বিনিয়োগ পরামর্শ নয়।",
),
)
self.last_evidence = {
"symbol": symbol,
"company": {
key: value
for key, value in company.items()
if key not in {"fifty_two_weeks_moving_range", "last_agm_date"}
},
"analysis_date": analysis_date.isoformat(),
"market_snapshot": analysis.market.model_dump(mode="json"),
"annual_financial_performance": financial[-10:],
"quarterly_performance": _rows_through(data["quarterly"], analysis_date)[-16:],
"balance_sheet_history": _balance_history_through(balance, cutoff_year),
"shareholding_history": _rows_through(data["shareholding"], analysis_date)[-12:],
"dividend_history": _rows_through(data["dividends"], analysis_date)[-10:],
"nav_history": _rows_through(data["nav"], analysis_date)[-10:],
"loan_status": data["loan"],
"operating_cash_flow_per_share_history": _rows_through(
data["nocfps"], analysis_date
)[-16:],
"price_momentum": momentum,
"technical_evidence": history.get("technical_evidence", {}),
"calculated_factors": [factor.model_dump(mode="json") for factor in factors],
"valuation_anchors": {
"current_price": valuation.current_price,
"methods": [method.model_dump(mode="json") for method in valuation.methods],
},
}
return analysis
@staticmethod
def _profit_factor(eps_values: list[float]) -> FactorCard:
......@@ -584,6 +708,8 @@ class StockAnalysisBuilder:
sector: str,
universe: list[dict[str, Any]],
cutoff_year: int,
*,
allow_live_peers: bool = True,
) -> tuple[ValuationSummary, FactorCard]:
del face_value
historical_pe = _median(
......@@ -591,7 +717,11 @@ class StockAnalysisBuilder:
)
current_pe = price / latest_eps if latest_eps and latest_eps > 0 else None
own_profit = latest_eps * historical_pe if latest_eps and historical_pe else None
peer_pe = self._peer_pe(symbol, sector, universe, cutoff_year) if sector else None
peer_pe = (
self._peer_pe(symbol, sector, universe, cutoff_year)
if sector and allow_live_peers
else None
)
peer_profit = latest_eps * peer_pe if latest_eps and peer_pe else None
historical_pb_values: list[float] = []
......@@ -729,21 +859,76 @@ class StockAnalysisBuilder:
)
@staticmethod
def _momentum(symbol: str, analysis_date: date) -> dict[str, float | bool | None]:
def _historical_snapshot(
symbol: str, analysis_date: date
) -> dict[str, Any]:
try:
frame = fetch_dse_ohlcv(
symbol,
(analysis_date - timedelta(days=180)).isoformat(),
(analysis_date - timedelta(days=370)).isoformat(),
analysis_date.isoformat(),
)
closes = [float(value) for value in frame["Close"].tail(60)]
if len(closes) < 2:
return {"return_30d": None, "quiet": False}
start = closes[-22] if len(closes) >= 22 else closes[0]
change = (closes[-1] / start - 1) * 100 if start else 0
return {"return_30d": round(change, 1), "quiet": abs(change) < 5}
closes = [float(value) for value in frame["Close"]]
if not closes:
return {}
latest = closes[-1]
previous = closes[-2] if len(closes) >= 2 else None
daily_change = latest - previous if previous else None
daily_percent = daily_change / previous * 100 if previous else None
momentum_start = closes[-22] if len(closes) >= 22 else closes[0]
momentum = (latest / momentum_start - 1) * 100 if momentum_start else None
raw_as_of = frame.iloc[-1].get("Date")
parsed_as_of = datetime.fromisoformat(str(raw_as_of)).replace(
tzinfo=timezone.utc
)
recent_bars = []
for _, row in frame.tail(60).iterrows():
recent_bars.append(
{
"date": str(row.get("Date"))[:10],
"open": _number(row.get("Open")),
"high": _number(row.get("High")),
"low": _number(row.get("Low")),
"close": _number(row.get("Close")),
"volume": _number(row.get("Volume")),
}
)
volumes = [
value
for value in (_number(raw) for raw in frame["Volume"].tail(20))
if value is not None
]
def moving_average(period: int) -> float | None:
if len(closes) < period:
return None
return round(sum(closes[-period:]) / period, 2)
return {
"latest_price": latest,
"previous_close": previous,
"change": round(daily_change, 2) if daily_change is not None else None,
"change_percent": (
round(daily_percent, 2) if daily_percent is not None else None
),
"fifty_two_week_low": min(closes),
"fifty_two_week_high": max(closes),
"as_of": parsed_as_of,
"return_30d": round(momentum, 1) if momentum is not None else None,
"quiet": abs(momentum) < 5 if momentum is not None else False,
"technical_evidence": {
"sma_20": moving_average(20),
"sma_50": moving_average(50),
"sma_200": moving_average(200),
"latest_volume": volumes[-1] if volumes else None,
"average_volume_20": (
round(sum(volumes) / len(volumes), 1) if volumes else None
),
"recent_bars": recent_bars,
},
}
except Exception:
return {"return_30d": None, "quiet": False}
return {}
@staticmethod
def _headline(score: int, momentum: dict[str, float | bool | None]) -> tuple[BilingualText, BilingualText]:
......
dohasecuritiesstockai/api/models.py
View file @ 63580797
......@@ -86,9 +86,34 @@ class AgentReports(APIModel):
raw_state: dict[str, Any] = Field(default_factory=dict)
class AITraderReport(APIModel):
rating: Literal["Buy", "Overweight", "Hold", "Underweight", "Sell"]
action: BilingualText
confidence: Literal["low", "medium", "high"]
executive_summary: BilingualText
investment_thesis: BilingualText
entry_strategy: BilingualText
risk_controls: BilingualText
catalysts: list[BilingualText] = Field(default_factory=list)
invalidation_conditions: list[BilingualText] = Field(default_factory=list)
time_horizon: BilingualText
class AIResearch(APIModel):
provider: str
model: str
mode: Literal["ai_fundamental", "multi_agent_synthesis"]
generated_at: datetime
score_confidence: Literal["low", "medium", "high"]
score_rationale: BilingualText
data_quality: BilingualText
valuation_method_weights: dict[str, float] = Field(default_factory=dict)
trader_report: AITraderReport
class EvidenceSource(APIModel):
name: str
source_type: Literal["dse_api", "agent_state", "calculation"]
source_type: Literal["dse_api", "agent_state", "calculation", "ai_analysis"]
detail: str
......@@ -111,6 +136,7 @@ class StockAnalysis(APIModel):
factors: list[FactorCard]
report_sections: list[ReportSection]
agent_reports: AgentReports
ai_research: AIResearch | None = None
sources: list[EvidenceSource]
disclaimer: BilingualText
......
dohasecuritiesstockai/api/prompts/__init__.py 0 → 100644
View file @ 63580797
"""Prompts used by the API-facing AI research workflow."""
dohasecuritiesstockai/api/prompts/stock_research.py 0 → 100644
View file @ 63580797
"""Grounded prompts for the one-call dashboard research synthesizer."""
from __future__ import annotations
AI_STOCK_RESEARCH_SYSTEM_PROMPT = """You are the senior Dhaka Stock Exchange (DSE)
equity analyst and portfolio-risk reviewer for Doha Securities Stock AI.
Your job is to turn the supplied, date-bounded evidence into a clear educational
research report and a disciplined trader view. You are not a data-retrieval agent:
use only the JSON evidence in the user message, do not browse, call tools, or rely on
unstated memory.
NON-NEGOTIABLE EVIDENCE RULES
1. Never invent a figure, company fact, event, audit opinion, catalyst, price level,
peer, or financial statement line. If evidence is absent, explicitly say it is
unavailable and lower confidence.
2. Respect analysis_date as the hard information cutoff. Never use later data.
3. Distinguish reported facts from your inference. Cite years/periods and figures in
the prose whenever the evidence contains them.
4. Treat zero live price during a closed market as unavailable; the evidence builder
may supply the positive previous close as today's usable reference price.
5. Do not treat interim dividends as a full-year dividend. Do not annualize incomplete
quarters unless the evidence explicitly provides a trailing or annual figure.
6. Do not confuse revenue, operating profit, net profit, EPS, NAV, NOCFPS, cash,
equity, debt, or market price. Preserve the units present in the evidence.
7. A qualified/adverse audit opinion, missing cash-flow evidence, unusual related-party
exposure, weak free float, or dependence on non-core income must be discussed only
when supplied evidence supports it.
8. The raw gateway does not declare the scaling unit for company paid-up capital,
reserve/surplus, the annual `profit` field, or columnar balance-sheet amounts.
Never label those raw amounts as crore, million, billion, lakh, or a BDT amount.
Prefer per-share figures; otherwise call them "gateway-reported units".
AI FUNDAMENTAL SCORE
Return 0-10 judgments for exactly five factors. The application, not you, calculates
the final 0-100 score using: profitability 25%, financial health 25%, business quality
20%, valuation 15%, and dividend quality 15%.
- Profitability: profitable-year record, EPS/profit trend, consistency, and earnings quality.
- Financial health: debt, liquidity/cash cushion, equity, and operating cash conversion.
- Business quality: durability, operating consistency, ownership/governance, and evidence
of a defensible position. Do not award an imagined moat.
- Valuation: current price relative to the four supplied valuation anchors and the
reliability/dispersion of those anchors.
- Dividend quality: payment consistency, growth, payout support, and current-price context.
Score 5 when evidence is mixed or materially incomplete; do not turn missing data into 0 or 10.
The application calculates and displays the final weighted 0-100 score after your
response. Never state, estimate, or repeat an overall `/100` score in any narrative
field. Discuss only your five 0-10 factor judgments and their evidence.
VALUE TODAY SYSTEM
The evidence contains four pre-calculated, auditable value anchors:
historical_pe, peer_pe, historical_pb, and dividend_yield. Do not alter those numeric
anchors. Assign each method a 0-1 reliability weight. Give zero weight to unavailable
methods; down-weight weak peer sets, unstable earnings, stale NAV, irregular dividends,
or extreme outliers. The application normalizes your weights, calculates the weighted
rough estimate, creates an educational fair range of 80%-120% of that estimate, and
derives Looks cheap/Fair/Looks expensive from current price. Your prose must explain
which methods deserve trust and why. This is not a price target.
The application performs that calculation after your response. Never state a weighted
rough estimate, fair-range endpoints, or final cheap/fair/expensive verdict in any
narrative field. You may quote the four immutable method anchors and current price.
FULL RESEARCH
Produce exactly ten sections: company, business_model, profitability,
financial_safety, valuation, dividends, moat, bull_case, risks, and suitability.
Make the report useful to a trader: include concrete yearly trends, data limitations,
catalysts, invalidation conditions, entry discipline, risk controls, and time horizon.
Never promise returns. Never personalize position sizing because the user's holdings,
risk tolerance, and liquidity needs are unknown.
For entry strategy, do not invent technical support/resistance or use the not-yet-known
weighted fair range. Use conditional staged-entry language and only quote an exact price
level if it is explicitly present in raw price history. Use volume or moving-average
claims only when `technical_evidence` supplies them. A 52-week low is an observed range
endpoint, not automatically proven support. Return at least two concrete
catalysts and two concrete invalidation conditions; if company-specific evidence is
missing, make the condition a future evidence check rather than inventing an event.
LANGUAGE AND STYLE
Every user-facing text field must contain both natural English (en) and natural Bangla
(bn), conveying the same facts. Use plain language first, with financial terminology
where it improves precision. Be decisive but calibrated. Do not use marketing language,
claim perfection, or present the output as investment advice.
"""
AI_STOCK_RESEARCH_TASK_TEMPLATE = """Analyze the supplied evidence for {symbol} as of
{analysis_date}. The current mode is {mode}. If mode is multi_agent_synthesis, reconcile
the supplied analyst/trader reports with the raw DSE evidence; raw numeric evidence wins
when prose conflicts with it. If mode is ai_fundamental, state that the trader view is
based on fundamentals, price history, and disclosures only—not a full multi-agent debate.
Return the required structured object. Keep each long section substantive but concise;
across the ten sections, cover every material strength, weakness, valuation issue, and
decision condition present in the evidence.
AUTHORITATIVE EVIDENCE JSON:
{evidence_json}
"""
dohasecuritiesstockai/api/service.py
View file @ 63580797
......@@ -11,6 +11,7 @@ from datetime import date, datetime, timezone
from dohasecuritiesstockai.default_config import DEFAULT_CONFIG
from dohasecuritiesstockai.graph.trading_graph import TradingAgentsGraph
from .ai_analysis import AIStockAnalysisGenerator
from .analysis import StockAnalysisBuilder
from .models import AnalysisJob, StockAnalysis
from .repository import AnalysisRepository
......@@ -62,7 +63,7 @@ class AnalysisService:
job = self._update(job, status="running", message="Collecting DSE evidence.")
if not force:
existing = self.repository.get_analysis(job.symbol, job.analysis_date)
if existing:
if existing and existing.ai_research is not None:
self._update(
job,
status="completed",
......@@ -90,10 +91,11 @@ class AnalysisService:
else:
state = self.repository.load_state(state_path)
job = self._update(job, message="Calculating the presentation score and valuation.")
analysis = StockAnalysisBuilder().build(
job.symbol, job.analysis_date, agent_state=state
job = self._update(
job,
message="Running the grounded AI score, valuation, and trader synthesis.",
)
analysis = self._build_analysis(job.symbol, job.analysis_date, state)
self.repository.save_analysis(analysis)
self._update(
job,
......@@ -119,6 +121,20 @@ class AnalysisService:
existing = self.repository.get_analysis(symbol, analysis_date)
if existing:
return existing
analysis = StockAnalysisBuilder().build(symbol, analysis_date, state)
analysis = self._build_analysis(symbol, analysis_date, state)
self.repository.save_analysis(analysis)
return analysis
@staticmethod
def _build_analysis(
symbol: str,
analysis_date: date,
state: dict,
) -> StockAnalysis:
builder = StockAnalysisBuilder()
analysis = builder.build(symbol, analysis_date, state)
return AIStockAnalysisGenerator(DEFAULT_CONFIG).enhance(
analysis,
builder.last_evidence,
state,
)
dohasecuritiesstockai/dashboard.py
View file @ 63580797
......@@ -20,6 +20,7 @@ import uvicorn
from dohasecuritiesstockai.api.analysis import StockAnalysisBuilder
from dohasecuritiesstockai.api.models import StockAnalysis
from dohasecuritiesstockai.api.repository import AnalysisRepository
from dohasecuritiesstockai.default_config import DEFAULT_CONFIG
class DashboardLaunchError(RuntimeError):
......@@ -29,17 +30,34 @@ class DashboardLaunchError(RuntimeError):
def prepare_dashboard_analysis(
symbol: str,
analysis_date: date | str,
agent_state: dict[str, Any],
agent_state: dict[str, Any] | None,
results_dir: str | Path,
*,
use_ai: bool = True,
config: dict[str, Any] | None = None,
) -> tuple[StockAnalysis, Path]:
"""Build and save the presentation model from the exact completed CLI state."""
"""Build and save the UI payload, optionally enriching it with agent state.
Passing ``agent_state=None`` deliberately skips the multi-agent graph. When
``use_ai`` is true, one configured model call synthesizes the already-fetched
evidence into the score, valuation weighting, report, and trader view.
"""
parsed_date = (
analysis_date
if isinstance(analysis_date, date)
else date.fromisoformat(str(analysis_date))
)
analysis = StockAnalysisBuilder().build(symbol, parsed_date, agent_state=agent_state)
builder = StockAnalysisBuilder()
analysis = builder.build(symbol, parsed_date, agent_state=agent_state)
if use_ai:
from dohasecuritiesstockai.api.ai_analysis import AIStockAnalysisGenerator
analysis = AIStockAnalysisGenerator(config or DEFAULT_CONFIG).enhance(
analysis,
builder.last_evidence,
agent_state,
)
path = AnalysisRepository(results_dir).save_analysis(analysis)
return analysis, path
......
dohasecuritiesstockai/dashboard_cli.py 0 → 100644
View file @ 63580797
"""Lightweight dashboard command that never starts the multi-agent graph."""
from __future__ import annotations
from datetime import date
from typing import Annotated
import typer
from rich.console import Console
from rich.panel import Panel
from dohasecuritiesstockai.dashboard import (
dashboard_url,
launch_dashboard,
prepare_dashboard_analysis,
)
from dohasecuritiesstockai.default_config import DEFAULT_CONFIG
console = Console()
def show_dashboard(
symbol: Annotated[
str,
typer.Argument(help="DSE symbol, for example GP or BRACBANK."),
],
analysis_date: Annotated[
str | None,
typer.Option("--date", "-d", help="Dashboard data date (YYYY-MM-DD)."),
] = None,
use_ai: Annotated[
bool,
typer.Option(
"--ai/--no-ai",
help="Use the configured AI for score, valuation weighting, and trader report.",
),
] = True,
open_ui: Annotated[
bool,
typer.Option(
"--open-ui/--no-open-ui",
help="Serve and open the dashboard after fetching its data.",
),
] = True,
host: Annotated[
str,
typer.Option(help="Local dashboard/API bind host."),
] = "127.0.0.1",
port: Annotated[
int,
typer.Option(min=1, max=65535, help="Local dashboard/API port."),
] = 8000,
) -> None:
"""Fetch only the DSE score-page data, then open the local dashboard."""
try:
requested_date = date.fromisoformat(analysis_date) if analysis_date else date.today()
if requested_date > date.today():
raise ValueError("--date cannot be in the future.")
except ValueError as exc:
console.print(f"[bold red]Invalid dashboard date:[/bold red] {exc}")
raise typer.Exit(code=2) from None
console.print(
Panel.fit(
f"[bold]DSE score dashboard[/bold]\n"
f"{symbol.strip().upper()} · {requested_date.isoformat()} · "
f"{'grounded AI research' if use_ai else 'calculated data only'}",
border_style="cyan",
)
)
try:
status = (
"Collecting yearly DSE evidence and running the configured AI analyst…"
if use_ai
else "Requesting the dashboard fields from the DSE data APIs…"
)
with console.status(status):
analysis, saved_path = prepare_dashboard_analysis(
symbol,
requested_date,
agent_state=None,
results_dir=DEFAULT_CONFIG["results_dir"],
use_ai=use_ai,
config=DEFAULT_CONFIG,
)
except Exception as exc:
console.print(f"[bold red]Dashboard data request failed:[/bold red] {exc}")
raise typer.Exit(code=1) from None
console.print(
f"[green]✓ Dashboard data saved:[/green] {saved_path}\n"
f"[green]✓ Fundamental score:[/green] "
f"{analysis.fundamental_score}/100 ({analysis.score_label.en})"
)
if analysis.ai_research is not None:
console.print(
f"[green]✓ AI analyst:[/green] {analysis.ai_research.provider} / "
f"{analysis.ai_research.model}\n"
f"[green]✓ Trader view:[/green] "
f"{analysis.ai_research.trader_report.rating}"
)
if not open_ui:
return
url = dashboard_url(host, port, analysis.symbol, analysis.analysis_date)
console.print(
f"[green]✓ Opening:[/green] {url}\n"
"[dim]Keep this terminal open; press Ctrl+C to stop the UI.[/dim]"
)
try:
launch_dashboard(
analysis.symbol,
analysis.analysis_date,
host=host,
port=port,
)
except Exception as exc:
console.print(f"[bold red]Dashboard launch failed:[/bold red] {exc}")
raise typer.Exit(code=1) from None
def main() -> None:
typer.run(show_dashboard)
if __name__ == "__main__":
main()
pyproject.toml
View file @ 63580797
......@@ -63,6 +63,7 @@ dohasecuritiesstockai = "cli.main:app"
tradingagents = "cli.main:app"
dohasecuritiesstockai-api = "dohasecuritiesstockai.api.__main__:main"
tradingagents-api = "dohasecuritiesstockai.api.__main__:main"
dohasecuritiesstockai-dashboard = "dohasecuritiesstockai.dashboard_cli:main"
dohasecuritiesstockai-predict = "dohasecuritiesstockai.timesfm_forecasting.cli:main"
tradingagents-predict = "dohasecuritiesstockai.timesfm_forecasting.cli:main"
dohasecuritiesstockai-dgt-predict = "dohasecuritiesstockai.dgt_forecasting.cli:main"
......
tests/test_ai_dashboard_analysis.py 0 → 100644
View file @ 63580797
from __future__ import annotations
from datetime import date, datetime, timezone
from types import SimpleNamespace
from dohasecuritiesstockai.api.ai_analysis import (
AIStockAnalysisGenerator,
AIStockResearchOutput,
_narrative_violations,
)
from dohasecuritiesstockai.api.models import (
AgentReports,
BilingualText,
EvidenceSource,
FactorCard,
MarketSnapshot,
ScoreMetric,
StockAnalysis,
ValuationMethod,
ValuationSummary,
)
def _text(value: str) -> dict[str, str]:
return {"en": value, "bn": f"বাংলা {value}"}
def _ai_output() -> AIStockResearchOutput:
factors = {
key: {"score": 8, "rationale": _text(f"{key} evidence")}
for key in (
"profitability",
"financial_health",
"business_quality",
"valuation",
"dividend",
)
}
section_keys = (
"company",
"business_model",
"profitability",
"financial_safety",
"valuation",
"dividends",
"moat",
"bull_case",
"risks",
"suitability",
)
return AIStockResearchOutput.model_validate(
{
"factors": factors,
"score_confidence": "high",
"score_rationale": _text("Grounded score rationale"),
"headline": _text("Strong evidence-backed profile"),
"takeaways": [_text("One"), _text("Two"), _text("Three")],
"in_depth_title": _text("AI in-depth research"),
"in_depth_summary": _text("Yearly evidence was reviewed."),
"valuation": {
"weights": {
"historical_pe": 1,
"peer_pe": 0,
"historical_pb": 0,
"dividend_yield": 0,
},
"confidence": "medium",
"summary": _text("The earnings-history method is most reliable."),
},
"sections": [
{
"key": key,
"title": _text(key.replace("_", " ").title()),
"summary": _text(f"Detailed {key} analysis"),
"bullets": [_text(f"{key} evidence point")],
}
for key in section_keys
],
"trader": {
"rating": "Overweight",
"action": _text("Accumulate only with entry discipline"),
"confidence": "medium",
"executive_summary": _text("Use staged entries and defined risk."),
"investment_thesis": _text("Fundamentals are constructive."),
"entry_strategy": _text("Wait for confirmation."),
"risk_controls": _text("Limit exposure and define invalidation."),
"catalysts": [
_text("Improving earnings"),
_text("Stable dividend coverage"),
],
"invalidation_conditions": [
_text("Earnings deterioration"),
_text("Dividend coverage weakens"),
],
"time_horizon": _text("Six to twelve months"),
},
"data_quality": _text("Annual evidence is available; some fields are missing."),
}
)
def _baseline() -> StockAnalysis:
labels = {
"profitability": "Profitability",
"financial_health": "Financial health",
"business_quality": "Business quality",
"valuation": "Valuation",
"dividend": "Dividend",
}
factors = [
FactorCard(
key=key,
status="caution",
title=BilingualText(en=label, bn=label),
subtitle=BilingualText(en="Baseline", bn="Baseline"),
explanation=BilingualText(en="Calculated", bn="Calculated"),
metrics=[
ScoreMetric(
key="calculated",
label=BilingualText(en="Calculated", bn="Calculated"),
display_value="5/10",
score=5,
)
],
)
for key, label in labels.items()
]
methods = [
ValuationMethod(
key=key,
label=BilingualText(en=key, bn=key),
value=value,
available=True,
)
for key, value in zip(
("historical_pe", "peer_pe", "historical_pb", "dividend_yield"),
(100, 200, 300, 400),
strict=True,
)
]
return StockAnalysis(
analysis_id="GP-2026-08-10",
symbol="GP",
company_name="Grameenphone Ltd.",
sector="Telecom",
analysis_date=date(2026, 8, 10),
generated_at=datetime.now(timezone.utc),
market=MarketSnapshot(
latest_price=50,
as_of=datetime.now(timezone.utc),
),
fundamental_score=50,
score_label=BilingualText(en="Mixed", bn="মিশ্র"),
headline=BilingualText(en="Baseline", bn="Baseline"),
takeaways=[],
in_depth_title=BilingualText(en="Baseline", bn="Baseline"),
in_depth_snippet=BilingualText(en="Baseline", bn="Baseline"),
valuation=ValuationSummary(
verdict="fair",
verdict_label=BilingualText(en="Fair price", bn="ন্যায্য দাম"),
current_price=50,
rough_estimate=250,
fair_range_low=200,
fair_range_high=300,
confidence="high",
summary=BilingualText(en="Calculated", bn="Calculated"),
methods=methods,
),
factors=factors,
report_sections=[],
agent_reports=AgentReports(),
sources=[
EvidenceSource(name="DSE", source_type="dse_api", detail="read-only")
],
disclaimer=BilingualText(en="Educational only", bn="শিক্ষামূলক"),
)
def test_ai_enhancement_recomputes_score_and_weighted_value() -> None:
captured: dict[str, object] = {}
output = _ai_output()
structured = SimpleNamespace(
invoke=lambda messages: captured.setdefault("messages", messages) and output
)
llm = SimpleNamespace(with_structured_output=lambda schema: structured)
generator = AIStockAnalysisGenerator(
{"llm_provider": "test", "deep_think_llm": "test-model"},
llm=llm,
)
result = generator.enhance(
_baseline(),
{"annual_financial_performance": [{"year": 2025, "eps_basic": "12.4"}]},
)
assert result.fundamental_score == 80
assert result.score_label.en == "Very good"
assert result.valuation.rough_estimate == 100
assert result.valuation.fair_range_low == 80
assert result.valuation.fair_range_high == 120
assert result.valuation.verdict == "looks_cheap"
assert result.ai_research is not None
assert result.ai_research.mode == "ai_fundamental"
assert result.ai_research.trader_report.rating == "Overweight"
assert len(result.report_sections) == 10
assert all(factor.metrics[0].key == "ai_factor_assessment" for factor in result.factors)
messages = captured["messages"]
assert "Never invent" in messages[0]["content"]
assert "Never state, estimate, or repeat an overall" in messages[0]["content"]
assert "Never state a weighted" in messages[0]["content"]
assert '"year": 2025' in messages[1]["content"]
assert "ai_fundamental" in messages[1]["content"]
assert '"fundamental_score"' not in messages[1]["content"]
def test_multi_agent_state_changes_ai_research_mode() -> None:
output = _ai_output()
structured = SimpleNamespace(invoke=lambda messages: output)
llm = SimpleNamespace(with_structured_output=lambda schema: structured)
generator = AIStockAnalysisGenerator(
{"llm_provider": "test", "deep_think_llm": "test-model"},
llm=llm,
)
result = generator.enhance(
_baseline(),
{},
{"final_trade_decision": "**Rating**: Hold"},
)
assert result.ai_research is not None
assert result.ai_research.mode == "multi_agent_synthesis"
def test_narrative_guard_retries_structured_output_once() -> None:
bad_payload = _ai_output().model_dump()
bad_payload["headline"] = {"en": "Old score 72/100", "bn": "পুরোনো ৭২/১০০"}
bad = AIStockResearchOutput.model_validate(bad_payload)
good = _ai_output()
responses = iter([bad, good])
calls: list[object] = []
def invoke(messages):
calls.append(messages)
return next(responses)
structured = SimpleNamespace(invoke=invoke)
llm = SimpleNamespace(with_structured_output=lambda schema: structured)
generator = AIStockAnalysisGenerator(
{"llm_provider": "test", "deep_think_llm": "test-model"},
llm=llm,
)
result = generator.enhance(_baseline(), {})
assert result.headline.en == good.headline.en
assert len(calls) == 2
assert _narrative_violations(good) == []
tests/test_analysis_api.py
View file @ 63580797
from datetime import date, datetime, timezone
from pathlib import Path
import pandas as pd
from fastapi.testclient import TestClient
from dohasecuritiesstockai.api.analysis import StockAnalysisBuilder
from dohasecuritiesstockai.api.analysis import (
StockAnalysisBuilder,
_market_price,
_rows_through,
)
from dohasecuritiesstockai.api.app import app
from dohasecuritiesstockai.api.models import AnalysisJob
from dohasecuritiesstockai.api.repository import AnalysisRepository
......@@ -39,6 +44,50 @@ def test_interim_dividend_is_not_treated_as_full_year() -> None:
assert result == {2025: 20.0}
def test_market_closed_quote_falls_back_to_previous_close() -> None:
quote = {"ltp": 0, "close": 0, "ycp": 258.7}
assert _market_price(quote) == 258.7
def test_ai_evidence_rows_respect_analysis_date() -> None:
rows = [
{"date": "2026-08-09", "eps_basic": "10"},
{"date": "2026-08-11", "eps_basic": "99"},
{"year": 2025, "eps_basic": "8"},
{"year": 2027, "eps_basic": "100"},
]
assert _rows_through(rows, date(2026, 8, 10)) == [rows[0], rows[2]]
def test_historical_snapshot_uses_last_candle_at_cutoff(monkeypatch) -> None:
frame = pd.DataFrame(
{
"Date": pd.to_datetime(["2026-08-09", "2026-08-10"]),
"Close": [256.6, 258.6],
"Open": [256.0, 257.0],
"High": [258.0, 259.0],
"Low": [255.5, 257.0],
"Volume": [160_000, 146_000],
}
)
monkeypatch.setattr(
"dohasecuritiesstockai.api.analysis.fetch_dse_ohlcv",
lambda symbol, start, end: frame,
)
snapshot = StockAnalysisBuilder._historical_snapshot("GP", date(2026, 8, 10))
assert snapshot["latest_price"] == 258.6
assert snapshot["previous_close"] == 256.6
assert snapshot["change_percent"] == 0.78
assert snapshot["as_of"] == datetime(2026, 8, 10, tzinfo=timezone.utc)
assert snapshot["technical_evidence"]["latest_volume"] == 146_000
assert snapshot["technical_evidence"]["average_volume_20"] == 153_000
assert snapshot["technical_evidence"]["recent_bars"][-1]["date"] == "2026-08-10"
def test_job_repository_round_trip(tmp_path: Path) -> None:
repository = AnalysisRepository(tmp_path)
now = datetime.now(timezone.utc)
......
tests/test_cli_branding.py
View file @ 63580797
......@@ -27,6 +27,10 @@ def test_product_command_and_legacy_alias_share_the_cli_entrypoint():
'dohasecuritiesstockai-api = "dohasecuritiesstockai.api.__main__:main"'
in project
)
assert (
'dohasecuritiesstockai-dashboard = "dohasecuritiesstockai.dashboard_cli:main"'
in project
)
def test_python_package_uses_dohasecuritiesstockai_directory():
......
tests/test_dashboard_flow.py
View file @ 63580797
from __future__ import annotations
import json
from datetime import date
from pathlib import Path
from types import SimpleNamespace
from fastapi.testclient import TestClient
import cli.main as cli_main
import dohasecuritiesstockai.dashboard_cli as dashboard_cli
from dohasecuritiesstockai.api.app import create_app
from dohasecuritiesstockai.dashboard import dashboard_url
from dohasecuritiesstockai.dashboard import dashboard_url, prepare_dashboard_analysis
from dohasecuritiesstockai.graph.trading_graph import TradingAgentsGraph
......@@ -17,6 +20,85 @@ def test_dashboard_url_targets_exact_cli_result() -> None:
)
def test_lightweight_dashboard_command_skips_agent_state(tmp_path: Path, monkeypatch) -> None:
calls: dict[str, object] = {}
analysis = SimpleNamespace(
symbol="GP",
analysis_date=date(2026, 8, 10),
fundamental_score=72,
score_label=SimpleNamespace(en="Good"),
ai_research=None,
)
def fake_prepare(
symbol, analysis_date, agent_state, results_dir, *, use_ai, config
):
calls["prepare"] = (
symbol,
analysis_date,
agent_state,
results_dir,
use_ai,
config,
)
return analysis, tmp_path / "GP.json"
def fake_launch(symbol, analysis_date, *, host, port):
calls["launch"] = (symbol, analysis_date, host, port)
monkeypatch.setattr(dashboard_cli, "prepare_dashboard_analysis", fake_prepare)
monkeypatch.setattr(dashboard_cli, "launch_dashboard", fake_launch)
monkeypatch.setitem(dashboard_cli.DEFAULT_CONFIG, "results_dir", str(tmp_path))
dashboard_cli.show_dashboard(
"gp",
analysis_date="2026-08-10",
use_ai=True,
open_ui=True,
host="0.0.0.0",
port=8123,
)
assert calls["prepare"] == (
"gp",
date(2026, 8, 10),
None,
str(tmp_path),
True,
dashboard_cli.DEFAULT_CONFIG,
)
assert calls["launch"] == ("GP", date(2026, 8, 10), "0.0.0.0", 8123)
def test_prepare_dashboard_analysis_accepts_lightweight_mode(tmp_path: Path, monkeypatch) -> None:
built = SimpleNamespace(symbol="GP", analysis_date=date(2026, 8, 10))
calls: dict[str, object] = {}
class FakeBuilder:
def build(self, symbol, analysis_date, agent_state):
calls["build"] = (symbol, analysis_date, agent_state)
return built
class FakeRepository:
def __init__(self, results_dir):
calls["results_dir"] = results_dir
def save_analysis(self, analysis):
calls["saved"] = analysis
return tmp_path / "analysis.json"
monkeypatch.setattr("dohasecuritiesstockai.dashboard.StockAnalysisBuilder", FakeBuilder)
monkeypatch.setattr("dohasecuritiesstockai.dashboard.AnalysisRepository", FakeRepository)
result, path = prepare_dashboard_analysis(
"GP", "2026-08-10", None, tmp_path, use_ai=False
)
assert result is built
assert path == tmp_path / "analysis.json"
assert calls["build"] == ("GP", date(2026, 8, 10), None)
def test_open_ui_env_flag_is_strict(monkeypatch) -> None:
monkeypatch.setenv("TRADINGAGENTS_OPEN_UI_AFTER_ANALYSIS", "yes")
assert cli_main._env_flag("TRADINGAGENTS_OPEN_UI_AFTER_ANALYSIS") is True
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment