How to Access Stock Analysis Results via the REST API in daily_stock_analysis

The ZhuLinsen/daily_stock_analysis repository exposes a complete REST API built with FastAPI that enables programmatic triggering of AI-driven stock analyses and retrieval of results through HTTP endpoints.

The stock analysis results API provides a production-ready interface for equity evaluation, supporting both real-time synchronous queries and asynchronous batch processing. Designed with a three-layer architecture, the system handles task queuing, duplicate detection, and persistent storage of completed reports in SQLite.

API Architecture and Design

The implementation follows a clean separation of concerns across routing, schema validation, and business logic layers.

Routing Layer

HTTP endpoints are registered under /api/v1/analysis in [api/v1/endpoints/analysis.py](https://github.com/ZhuLinsen/daily_stock_analysis/blob/main/api/v1/endpoints/analysis.py). This module maps incoming requests to service functions and handles path operations for analysis triggering, status checking, task listing, and real-time streaming.

Schema Layer

Request and response models are defined using Pydantic in [api/v1/schemas/analysis.py](https://github.com/ZhuLinsen/daily_stock_analysis/blob/main/api/v1/schemas/analysis.py). Key models include:

• AnalyzeRequest – Validates input parameters including stock_code, stock_codes (batch), async_mode, and report_type.
• AnalysisResultResponse – Wraps the complete analysis report with metadata.
• TaskStatus – Tracks execution state, progress percentages, and embedded results.
• TaskListResponse – Provides paginated task summaries with filter support.

The report structure itself conforms to the AnalysisReport schema defined in [api/v1/schemas/history.py](https://github.com/ZhuLinsen/daily_stock_analysis/blob/main/api/v1/schemas/history.py).

Service Layer

Core business logic resides in [src/services/analysis_service.py](https://github.com/ZhuLinsen/daily_stock_analysis/blob/main/src/services/analysis_service.py), which executes AI-driven analysis synchronously or enqueues asynchronous jobs. The TaskQueue implementation in [src/services/task_queue.py](https://github.com/ZhuLinsen/daily_stock_analysis/blob/main/src/services/task_queue.py) manages in-memory task scheduling and duplicate detection, returning HTTP 409 Conflict for duplicate submissions.

Core API Workflow

The API supports a complete lifecycle from request submission to result retrieval.

Triggering Analysis

POST /api/v1/analysis/analyze accepts single or batch requests. For single-stock synchronous analysis, set async_mode: false; the endpoint returns AnalysisResultResponse immediately. For batch processing, set async_mode: true to receive a BatchTaskAcceptedResponse (HTTP 202 Accepted) containing task IDs for queue tracking.

Checking Task Status

GET /api/v1/analysis/status/{task_id} queries the TaskQueue or SQLite database via [src/storage/DatabaseManager.py](https://github.com/ZhuLinsen/daily_stock_analysis/blob/main/src/storage/DatabaseManager.py). Returns TaskStatus with current progress; if completed, the result field contains the full AnalysisResultResponse.

Listing Historical Tasks

GET /api/v1/analysis/tasks supports optional filtering by status (e.g., completed,pending) and pagination. Returns a TaskListResponse with total counts and TaskInfo objects for audit trails.

Real-Time Updates via SSE

GET /api/v1/analysis/tasks/stream opens a Server-Sent Events connection that pushes task_created, progress, task_completed, and heartbeat events. This enables live frontend updates without polling overhead.

Code Examples

The following examples assume the server is running locally on port 8000 via uvicorn as defined in [server.py](https://github.com/ZhuLinsen/daily_stock_analysis/blob/main/server.py).

Synchronous Single-Stock Analysis

Trigger immediate analysis for one stock code:

curl -X POST "http://localhost:8000/api/v1/analysis/analyze" \
  -H "Content-Type: application/json" \
  -d '{
        "stock_code": "600519",
        "async_mode": false,
        "report_type": "detailed"
      }'

Response (200): Returns an AnalysisResultResponse containing the query_id, stock_name, and full report structure generated by _build_analysis_report.

Asynchronous Batch Analysis

Submit multiple stocks for background processing:

curl -X POST "http://localhost:8000/api/v1/analysis/analyze" \
  -H "Content-Type: application/json" \
  -d '{
        "stock_codes": ["AAPL", "TSLA", "600519"],
        "async_mode": true,
        "report_type": "summary"
      }'

Response (202): Returns task IDs for accepted jobs; duplicate codes in the batch receive 409 conflict details in the response body.

Querying Task Status

Poll for completion using the task ID returned from an async submission:

TASK_ID="abcd1234efgh5678"
curl "http://localhost:8000/api/v1/analysis/status/${TASK_ID}"

• 200 OK: TaskStatus object with "status": "completed" and populated result field.
• 404 Not Found: Task ID does not exist or expired from queue.

Listing Tasks with Filters

Retrieve the ten most recent completed tasks:

curl "http://localhost:8000/api/v1/analysis/tasks?status=completed&limit=10"

Returns aggregated statistics including total, pending, and processing counts alongside the task list.

Subscribing to Real-Time Events

Connect to the SSE endpoint for live updates:

curl -N "http://localhost:8000/api/v1/analysis/tasks/stream"

The stream emits events formatted as:

event: task_created
data: {"task_id":"...","stock_code":"AAPL","status":"pending"}

event: heartbeat
data: {"timestamp":"2026-04-30T12:34:56.789Z"}

Client implementations should parse the event type and data payload accordingly.

Key Source Files

File	Responsibility
[api/v1/endpoints/analysis.py](https://github.com/ZhuLinsen/daily_stock_analysis/blob/main/api/v1/endpoints/analysis.py)	HTTP route definitions for /analyze, /status, /tasks, and /tasks/stream.
[api/v1/schemas/analysis.py](https://github.com/ZhuLinsen/daily_stock_analysis/blob/main/api/v1/schemas/analysis.py)	Pydantic models for request validation and response serialization.
[api/v1/schemas/history.py](https://github.com/ZhuLinsen/daily_stock_analysis/blob/main/api/v1/schemas/history.py)	AnalysisReport schema definitions including ReportMeta and ReportSummary.
[src/services/analysis_service.py](https://github.com/ZhuLinsen/daily_stock_analysis/blob/main/src/services/analysis_service.py)	AI analysis execution logic for synchronous calls.
[src/services/task_queue.py](https://github.com/ZhuLinsen/daily_stock_analysis/blob/main/src/services/task_queue.py)	In-memory queue management and duplicate detection.
[src/storage/DatabaseManager.py](https://github.com/ZhuLinsen/daily_stock_analysis/blob/main/src/storage/DatabaseManager.py)	SQLite persistence layer for completed reports.
[api/v1/router.py](https://github.com/ZhuLinsen/daily_stock_analysis/blob/main/api/v1/router.py)	Version 1 API router registration.
[api/app.py](https://github.com/ZhuLinsen/daily_stock_analysis/blob/main/api/app.py)	FastAPI application factory and middleware configuration.

Summary

• The stock analysis results API is implemented with FastAPI and follows a strict three-layer architecture separating routing, schemas, and service concerns.
• Endpoints support both synchronous immediate response mode and asynchronous batch processing with task queue management.
• Real-time status updates are available via Server-Sent Events at /api/v1/analysis/tasks/stream.
• Completed analysis reports persist in SQLite through DatabaseManager and remain retrievable via the status endpoint indefinitely.
• Duplicate submission protection returns HTTP 409 Conflict to prevent redundant processing.

Frequently Asked Questions

Is there an API for stock analysis results?

Yes. The repository exposes a full REST API under the /api/v1/analysis path prefix, implemented in [api/v1/endpoints/analysis.py](https://github.com/ZhuLinsen/daily_stock_analysis/blob/main/api/v1/endpoints/analysis.py) using FastAPI. This API accepts stock codes, triggers AI-driven evaluations, and returns structured JSON reports containing fundamental and technical analysis data.

How do I check if a stock analysis task is complete?

Send a GET request to /api/v1/analysis/status/{task_id} using the UUID returned when you triggered the analysis. The response follows the TaskStatus schema: if "status": "completed", the result field contains the AnalysisResultResponse with the full report; otherwise, check the progress percentage for queue position.

Can I analyze multiple stocks in one API call?

Yes. Submit a POST request to /api/v1/analysis/analyze with the stock_codes array parameter and async_mode: true. The service enqueues each stock as an independent task and returns a BatchTaskAcceptedResponse containing all task IDs. Duplicate stock codes within the batch are rejected with 409 status details while valid codes proceed normally.

Where are analysis results stored?

Completed reports are persisted to a SQLite database via [src/storage/DatabaseManager.py](https://github.com/ZhuLinsen/daily_stock_analysis/blob/main/src/storage/DatabaseManager.py). When querying task status, the API first checks the in-memory TaskQueue for active jobs; if the task is historic, it loads the report from SQLite, ensuring data durability across application restarts.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}