#!/usr/bin/env python # -*- coding: utf-8 -*- from fastapi import FastAPI, HTTPException, Query from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel from typing import List, Dict, Any, Optional from datetime import datetime, timedelta import logging from models import FinancialDataManager, NewsManager, AnalyticsManager from config import API_CONFIG # Настройка логирования logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler("api.log"), logging.StreamHandler() ] ) logger = logging.getLogger("api") # Создаем экземпляр FastAPI app = FastAPI( title="Финансово-аналитическая платформа API", description="API для доступа к финансовым данным, новостям и аналитике", version="1.0.0" ) # Настройка CORS app.add_middleware( CORSMiddleware, allow_origins=["*"], # В продакшене следует указать конкретные домены allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # Инициализация менеджеров данных financial_manager = FinancialDataManager() news_manager = NewsManager() analytics_manager = AnalyticsManager() # Модели данных для API class SectorData(BaseModel): sector: str date: str open: float high: float low: float close: float volume: float class TickerData(BaseModel): ticker: str sector: str date: str open: float high: float low: float close: float volume: float class NewsItem(BaseModel): id: int date: str topic: Optional[str] = None title: str content: str url: Optional[str] = None class AnalyticsItem(BaseModel): id: int date: str summary: str business_problems: Optional[str] = None solutions: Optional[str] = None sentiment: Optional[float] = None # Маршруты API @app.get("/") async def root(): """Корневой маршрут API""" return { "message": "Добро пожаловать в API финансово-аналитической платформы", "version": "1.0.0", "endpoints": [ "/sectors", "/sectors/{sector}", "/tickers", "/tickers/{ticker}", "/news", "/news/topics", "/analytics" ] } # Маршруты для финансовых данных @app.get("/sectors", response_model=List[str]) async def get_sectors(): """Получить список всех доступных секторов""" try: sectors = financial_manager.get_all_sectors() return sectors except Exception as e: logger.error(f"Ошибка при получении списка секторов: {str(e)}") raise HTTPException(status_code=500, detail=f"Внутренняя ошибка сервера: {str(e)}") @app.get("/sectors/{sector}", response_model=List[Dict[str, Any]]) async def get_sector_data( sector: str, start_date: Optional[str] = Query(None, description="Начальная дата в формате YYYY-MM-DD"), end_date: Optional[str] = Query(None, description="Конечная дата в формате YYYY-MM-DD") ): """Получить данные по указанному сектору за период""" try: # Если даты не указаны, используем последние 30 дней if not end_date: end_date = datetime.now().strftime("%Y-%m-%d") if not start_date: start_date = (datetime.now() - timedelta(days=30)).strftime("%Y-%m-%d") data = financial_manager.get_sector_data(sector, start_date, end_date) if not data: raise HTTPException(status_code=404, detail=f"Данные для сектора {sector} не найдены") return data except HTTPException: raise except Exception as e: logger.error(f"Ошибка при получении данных сектора {sector}: {str(e)}") raise HTTPException(status_code=500, detail=f"Внутренняя ошибка сервера: {str(e)}") @app.get("/tickers", response_model=Dict[str, List[str]]) async def get_tickers(): """Получить список всех доступных тикеров, сгруппированных по секторам""" try: tickers = financial_manager.get_all_tickers() return tickers except Exception as e: logger.error(f"Ошибка при получении списка тикеров: {str(e)}") raise HTTPException(status_code=500, detail=f"Внутренняя ошибка сервера: {str(e)}") @app.get("/tickers/{ticker}", response_model=List[Dict[str, Any]]) async def get_ticker_data( ticker: str, start_date: Optional[str] = Query(None, description="Начальная дата в формате YYYY-MM-DD"), end_date: Optional[str] = Query(None, description="Конечная дата в формате YYYY-MM-DD") ): """Получить данные по указанному тикеру за период""" try: # Если даты не указаны, используем последние 30 дней if not end_date: end_date = datetime.now().strftime("%Y-%m-%d") if not start_date: start_date = (datetime.now() - timedelta(days=30)).strftime("%Y-%m-%d") data = financial_manager.get_ticker_data(ticker, start_date, end_date) if not data: raise HTTPException(status_code=404, detail=f"Данные для тикера {ticker} не найдены") return data except HTTPException: raise except Exception as e: logger.error(f"Ошибка при получении данных тикера {ticker}: {str(e)}") raise HTTPException(status_code=500, detail=f"Внутренняя ошибка сервера: {str(e)}") # Маршруты для новостей @app.get("/news", response_model=List[Dict[str, Any]]) async def get_news( start_date: Optional[str] = Query(None, description="Начальная дата в формате YYYY-MM-DD"), end_date: Optional[str] = Query(None, description="Конечная дата в формате YYYY-MM-DD"), topic: Optional[str] = Query(None, description="Фильтр по теме новости") ): """Получить новости за указанный период с возможностью фильтрации по теме""" try: # Если даты не указаны, используем последние 7 дней if not end_date: end_date = datetime.now().strftime("%Y-%m-%d") if not start_date: start_date = (datetime.now() - timedelta(days=7)).strftime("%Y-%m-%d") if topic: news = news_manager.get_news_by_topic(topic, start_date, end_date) else: news = news_manager.get_news_by_date(start_date, end_date) return news except Exception as e: logger.error(f"Ошибка при получении новостей: {str(e)}") raise HTTPException(status_code=500, detail=f"Внутренняя ошибка сервера: {str(e)}") @app.get("/news/topics", response_model=List[str]) async def get_news_topics(): """Получить список всех доступных тем новостей""" try: topics = news_manager.get_topics() return topics except Exception as e: logger.error(f"Ошибка при получении списка тем новостей: {str(e)}") raise HTTPException(status_code=500, detail=f"Внутренняя ошибка сервера: {str(e)}") # Маршруты для аналитики @app.get("/analytics", response_model=List[Dict[str, Any]]) async def get_analytics( start_date: Optional[str] = Query(None, description="Начальная дата в формате YYYY-MM-DD"), end_date: Optional[str] = Query(None, description="Конечная дата в формате YYYY-MM-DD") ): """Получить аналитику за указанный период""" try: # Если даты не указаны, используем последние 7 дней if not end_date: end_date = datetime.now().strftime("%Y-%m-%d") if not start_date: start_date = (datetime.now() - timedelta(days=7)).strftime("%Y-%m-%d") analytics = analytics_manager.get_analytics_by_date(start_date, end_date) return analytics except Exception as e: logger.error(f"Ошибка при получении аналитики: {str(e)}") raise HTTPException(status_code=500, detail=f"Внутренняя ошибка сервера: {str(e)}") # Запуск сервера if __name__ == "__main__": import uvicorn uvicorn.run( "api:app", host=API_CONFIG['host'], port=API_CONFIG['port'], reload=True )