from sqlalchemy.orm import Session from fastapi import HTTPException, status from typing import List, Optional from datetime import datetime, date from decimal import Decimal from . import repository from .schema import ( SectorCreate, SectorUpdate, SectorResponse, ExchangeCreate, ExchangeUpdate, ExchangeResponse, StockCreate, StockUpdate, StockResponse, StockDetailResponse, StockhistoryCreate, StockhistoryUpdate, StockhistoryResponse, StockWithHistoryResponse, BulkImportStocksRequest, BulkImportStocksResponse, BulkImportHistoryRequest, BulkImportHistoryResponse ) def _get_or_raise(db: Session, entity_id: str, repo_module, entity_name: str = "Entity"): entity = repo_module.get_by_id(db, entity_id) if not entity: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"{entity_name} with id {entity_id} not found" ) return entity # Sector handlers def create_sector(db: Session, data: SectorCreate) -> SectorResponse: existing = repository.get_sector_by_name(db, data.name) if existing: raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail=f"Sector with name '{data.name}' already exists" ) try: sector = repository.create(db, data.model_dump()) db.commit() db.refresh(sector) return SectorResponse.model_validate(sector) except Exception: db.rollback() raise def list_sectors(db: Session, limit: int, offset: int, search: Optional[str] = None) -> List[SectorResponse]: filters = {} if search: filters["search"] = search sectors = repository.list_all(db, limit, offset, **filters) return [SectorResponse.model_validate(s) for s in sectors] def get_sector(db: Session, sector_id: str) -> SectorResponse: sector = _get_or_raise(db, sector_id, repository, "Sector") return SectorResponse.model_validate(sector) def update_sector(db: Session, sector_id: str, data: SectorUpdate) -> SectorResponse: sector = _get_or_raise(db, sector_id, repository, "Sector") update_data = data.model_dump(exclude_unset=True) if "name" in update_data and update_data["name"] != sector.name: existing = repository.get_sector_by_name(db, update_data["name"]) if existing: raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail=f"Sector with name '{update_data['name']}' already exists" ) try: updated_sector = repository.update(db, sector_id, update_data) db.commit() db.refresh(updated_sector) return SectorResponse.model_validate(updated_sector) except Exception: db.rollback() raise def delete_sector(db: Session, sector_id: str) -> dict: sector = _get_or_raise(db, sector_id, repository, "Sector") try: repository.delete(db, sector_id) db.commit() return {"message": "Sector deleted successfully"} except Exception: db.rollback() raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail="Cannot delete sector because it is referenced by existing stocks" ) # Exchange handlers def create_exchange(db: Session, data: ExchangeCreate) -> ExchangeResponse: existing = repository.get_exchange_by_code(db, data.code) if existing: raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail=f"Exchange with code '{data.code}' already exists" ) try: exchange = repository.create_exchange(db, data.model_dump()) db.commit() db.refresh(exchange) return ExchangeResponse.model_validate(exchange) except Exception: db.rollback() raise def list_exchanges(db: Session, limit: int, offset: int, search: Optional[str] = None, country: Optional[str] = None) -> List[ExchangeResponse]: filters = {} if search: filters["search"] = search if country: filters["country"] = country exchanges = repository.list_exchanges(db, limit, offset, **filters) return [ExchangeResponse.model_validate(e) for e in exchanges] def get_exchange(db: Session, exchange_id: str) -> ExchangeResponse: exchange = repository.get_exchange_by_id(db, exchange_id) if not exchange: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Exchange with id {exchange_id} not found" ) return ExchangeResponse.model_validate(exchange) def update_exchange(db: Session, exchange_id: str, data: ExchangeUpdate) -> ExchangeResponse: exchange = repository.get_exchange_by_id(db, exchange_id) if not exchange: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Exchange with id {exchange_id} not found" ) update_data = data.model_dump(exclude_unset=True) if "code" in update_data and update_data["code"] != exchange.code: existing = repository.get_exchange_by_code(db, update_data["code"]) if existing: raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail=f"Exchange with code '{update_data['code']}' already exists" ) try: updated_exchange = repository.update_exchange(db, exchange_id, update_data) db.commit() db.refresh(updated_exchange) return ExchangeResponse.model_validate(updated_exchange) except Exception: db.rollback() raise def delete_exchange(db: Session, exchange_id: str) -> dict: exchange = repository.get_exchange_by_id(db, exchange_id) if not exchange: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Exchange with id {exchange_id} not found" ) try: repository.delete_exchange(db, exchange_id) db.commit() return {"message": "Exchange deleted successfully"} except Exception: db.rollback() raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail="Cannot delete exchange because it is referenced by existing stocks" ) # Stock handlers def create_stock(db: Session, data: StockCreate) -> StockResponse: existing_ticker = repository.get_stock_by_ticker(db, data.ticker_symbol) if existing_ticker: raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail=f"Stock with ticker symbol '{data.ticker_symbol}' already exists" ) existing_company = repository.get_stock_by_company_name(db, data.company_name) if existing_company: raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail=f"Stock with company name '{data.company_name}' already exists" ) sector = repository.get_by_id(db, data.sector_id) if not sector: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"Sector with id {data.sector_id} does not exist" ) exchange = repository.get_exchange_by_id(db, data.exchange_id) if not exchange: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"Exchange with id {data.exchange_id} does not exist" ) if data.high_price is not None and data.low_price is not None: if data.high_price < data.low_price: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="High price must be greater than or equal to low price" ) if data.opening_price is not None and data.high_price is not None and data.low_price is not None: if data.opening_price < data.low_price or data.opening_price > data.high_price: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Opening price must be between low and high price" ) if data.closing_price is not None and data.high_price is not None and data.low_price is not None: if data.closing_price < data.low_price or data.closing_price > data.high_price: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Closing price must be between low and high price" ) try: stock = repository.create_stock(db, data.model_dump()) db.commit() db.refresh(stock) return StockResponse.model_validate(stock) except Exception: db.rollback() raise def list_stocks( db: Session, limit: int, offset: int, search: Optional[str] = None, sector_id: Optional[str] = None, exchange_id: Optional[str] = None, status_filter: Optional[str] = None, min_price: Optional[Decimal] = None, max_price: Optional[Decimal] = None ) -> List[StockResponse]: filters = {} if search: filters["search"] = search if sector_id: filters["sector_id"] = sector_id if exchange_id: filters["exchange_id"] = exchange_id if status_filter: filters["status"] = status_filter if min_price is not None: filters["min_price"] = min_price if max_price is not None: filters["max_price"] = max_price stocks = repository.list_stocks(db, limit, offset, **filters) return [StockResponse.model_validate(s) for s in stocks] def get_stock(db: Session, stock_id: str) -> StockResponse: stock = repository.get_stock_by_id(db, stock_id) if not stock: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Stock with id {stock_id} not found" ) return StockResponse.model_validate(stock) def get_stock_details(db: Session, stock_id: str) -> StockDetailResponse: stock = repository.get_stock_with_details(db, stock_id) if not stock: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Stock with id {stock_id} not found" ) return StockDetailResponse.model_validate(stock) def get_stock_with_history(db: Session, stock_id: str) -> StockWithHistoryResponse: stock = repository.get_stock_with_history(db, stock_id) if not stock: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Stock with id {stock_id} not found" ) return StockWithHistoryResponse.model_validate(stock) def search_stocks_autocomplete(db: Session, search_term: str) -> List[StockResponse]: if not search_term or len(search_term.strip()) < 1: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Search term must be at least 1 character" ) stocks = repository.search_stocks_autocomplete(db, search_term.strip(), limit=10) return [StockResponse.model_validate(s) for s in stocks] def update_stock(db: Session, stock_id: str, data: StockUpdate) -> StockResponse: stock = repository.get_stock_by_id(db, stock_id) if not stock: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Stock with id {stock_id} not found" ) update_data = data.model_dump(exclude_unset=True) if "ticker_symbol" in update_data and update_data["ticker_symbol"] != stock.ticker_symbol: existing = repository.get_stock_by_ticker(db, update_data["ticker_symbol"]) if existing: raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail=f"Stock with ticker symbol '{update_data['ticker_symbol']}' already exists" ) if "company_name" in update_data and update_data["company_name"] != stock.company_name: existing = repository.get_stock_by_company_name(db, update_data["company_name"]) if existing: raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail=f"Stock with company name '{update_data['company_name']}' already exists" ) if "sector_id" in update_data: sector = repository.get_by_id(db, update_data["sector_id"]) if not sector: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"Sector with id {update_data['sector_id']} does not exist" ) if "exchange_id" in update_data: exchange = repository.get_exchange_by_id(db, update_data["exchange_id"]) if not exchange: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"Exchange with id {update_data['exchange_id']} does not exist" ) high_price = update_data.get("high_price", stock.high_price) low_price = update_data.get("low_price", stock.low_price) if high_price is not None and low_price is not None: if high_price < low_price: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="High price must be greater than or equal to low price" ) opening_price = update_data.get("opening_price", stock.opening_price) if opening_price is not None and high_price is not None and low_price is not None: if opening_price < low_price or opening_price > high_price: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Opening price must be between low and high price" ) closing_price = update_data.get("closing_price", stock.closing_price) if closing_price is not None and high_price is not None and low_price is not None: if closing_price < low_price or closing_price > high_price: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Closing price must be between low and high price" ) update_data["last_updated"] = datetime.utcnow() try: updated_stock = repository.update_stock(db, stock_id, update_data) db.commit() db.refresh(updated_stock) return StockResponse.model_validate(updated_stock) except Exception: db.rollback() raise def delete_stock(db: Session, stock_id: str) -> dict: stock = repository.get_stock_by_id(db, stock_id) if not stock: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Stock with id {stock_id} not found" ) try: stock.status = "INACTIVE" db.commit() return {"message": "Stock marked as inactive successfully"} except Exception: db.rollback() raise def bulk_import_stocks(db: Session, data: BulkImportStocksRequest) -> BulkImportStocksResponse: total = len(data.stocks) successful = 0 failed = 0 errors = [] sector_names = list({item.sector_name for item in data.stocks}) sectors_list = db.query(repository.Sector).filter(repository.Sector.name.in_(sector_names)).all() if sector_names else [] sectors_map = {s.name: s for s in sectors_list} exchange_codes = list({item.exchange_code for item in data.stocks}) exchanges_list = db.query(repository.Exchange).filter(repository.Exchange.code.in_(exchange_codes)).all() if exchange_codes else [] exchanges_map = {e.code: e for e in exchanges_list} for idx, item in enumerate(data.stocks): try: sector = sectors_map.get(item.sector_name) if not sector: errors.append(f"Row {idx + 1}: Sector '{item.sector_name}' not found") failed += 1 continue exchange = exchanges_map.get(item.exchange_code) if not exchange: errors.append(f"Row {idx + 1}: Exchange '{item.exchange_code}' not found") failed += 1 continue existing_ticker = repository.get_stock_by_ticker(db, item.ticker_symbol) if existing_ticker: errors.append(f"Row {idx + 1}: Ticker '{item.ticker_symbol}' already exists") failed += 1 continue existing_company = repository.get_stock_by_company_name(db, item.company_name) if existing_company: errors.append(f"Row {idx + 1}: Company '{item.company_name}' already exists") failed += 1 continue stock_data = { "ticker_symbol": item.ticker_symbol, "company_name": item.company_name, "sector_id": sector.id, "exchange_id": exchange.id, "currency": item.currency, "current_price": item.current_price, "industry": item.industry, "description": item.description, "country": item.country, "ipo_date": item.ipo_date, "status": "ACTIVE", } repository.create_stock(db, stock_data) successful += 1 except Exception as e: errors.append(f"Row {idx + 1}: {str(e)}") failed += 1 try: db.commit() except Exception as e: db.rollback() raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Failed to commit bulk import: {str(e)}" ) return BulkImportStocksResponse( total=total, successful=successful, failed=failed, errors=errors ) # StockHistory handlers def create_stockhistory(db: Session, data: StockhistoryCreate) -> StockhistoryResponse: stock = repository.get_stock_by_id(db, data.stock_id) if not stock: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"Stock with id {data.stock_id} does not exist" ) existing = repository.get_stockhistory_by_stock_and_date(db, data.stock_id, data.date) if existing: raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail=f"Stock history for stock {data.stock_id} on date {data.date} already exists" ) if data.high_price < data.low_price: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="High price must be greater than or equal to low price" ) if data.open_price < data.low_price or data.open_price > data.high_price: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Opening price must be between low and high price" ) if data.close_price < data.low_price or data.close_price > data.high_price: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Closing price must be between low and high price" ) try: stockhistory = repository.create_stockhistory(db, data.model_dump()) db.commit() db.refresh(stockhistory) return StockhistoryResponse.model_validate(stockhistory) except Exception: db.rollback() raise def list_stockhistory( db: Session, limit: int, offset: int, stock_id: Optional[str] = None, start_date: Optional[date] = None, end_date: Optional[date] = None ) -> List[StockhistoryResponse]: filters = {} if stock_id: filters["stock_id"] = stock_id if start_date: filters["start_date"] = start_date if end_date: filters["end_date"] = end_date history = repository.list_stockhistory(db, limit, offset, **filters) return [StockhistoryResponse.model_validate(h) for h in history] def get_stockhistory(db: Session, stockhistory_id: str) -> StockhistoryResponse: stockhistory = repository.get_stockhistory_by_id(db, stockhistory_id) if not stockhistory: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Stock history with id {stockhistory_id} not found" ) return StockhistoryResponse.model_validate(stockhistory) def update_stockhistory(db: Session, stockhistory_id: str, data: StockhistoryUpdate) -> StockhistoryResponse: stockhistory = repository.get_stockhistory_by_id(db, stockhistory_id) if not stockhistory: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Stock history with id {stockhistory_id} not found" ) update_data = data.model_dump(exclude_unset=True) if "stock_id" in update_data: stock = repository.get_stock_by_id(db, update_data["stock_id"]) if not stock: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"Stock with id {update_data['stock_id']} does not exist" ) if "date" in update_data and update_data["date"] != stockhistory.date: stock_id = update_data.get("stock_id", stockhistory.stock_id) existing = repository.get_stockhistory_by_stock_and_date(db, stock_id, update_data["date"]) if existing: raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail=f"Stock history for stock {stock_id} on date {update_data['date']} already exists" ) high_price = update_data.get("high_price", stockhistory.high_price) low_price = update_data.get("low_price", stockhistory.low_price) if high_price < low_price: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="High price must be greater than or equal to low price" ) open_price = update_data.get("open_price", stockhistory.open_price) if open_price < low_price or open_price > high_price: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Opening price must be between low and high price" ) close_price = update_data.get("close_price", stockhistory.close_price) if close_price < low_price or close_price > high_price: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Closing price must be between low and high price" ) try: updated_stockhistory = repository.update_stockhistory(db, stockhistory_id, update_data) db.commit() db.refresh(updated_stockhistory) return StockhistoryResponse.model_validate(updated_stockhistory) except Exception: db.rollback() raise def delete_stockhistory(db: Session, stockhistory_id: str) -> dict: stockhistory = repository.get_stockhistory_by_id(db, stockhistory_id) if not stockhistory: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Stock history with id {stockhistory_id} not found" ) try: repository.delete_stockhistory(db, stockhistory_id) db.commit() return {"message": "Stock history deleted successfully"} except Exception: db.rollback() raise def bulk_import_stock_history(db: Session, data: BulkImportHistoryRequest) -> BulkImportHistoryResponse: total = len(data.history_items) successful = 0 failed = 0 errors = [] ticker_symbols = list({item.ticker_symbol for item in data.history_items}) stocks_list = db.query(repository.Stock).filter(repository.Stock.ticker_symbol.in_(ticker_symbols)).all() if ticker_symbols else [] stocks_map = {s.ticker_symbol: s for s in stocks_list} for idx, item in enumerate(data.history_items): try: stock = stocks_map.get(item.ticker_symbol) if not stock: errors.append(f"Row {idx + 1}: Stock with ticker '{item.ticker_symbol}' not found") failed += 1 continue existing = repository.get_stockhistory_by_stock_and_date(db, stock.id, item.date) if existing: errors.append(f"Row {idx + 1}: History for ticker '{item.ticker_symbol}' on {item.date} already exists") failed += 1 continue if item.high_price < item.low_price: errors.append(f"Row {idx + 1}: High price must be >= low price") failed += 1 continue if item.open_price < item.low_price or item.open_price > item.high_price: errors.append(f"Row {idx + 1}: Open price must be between low and high") failed += 1 continue if item.close_price < item.low_price or item.close_price > item.high_price: errors.append(f"Row {idx + 1}: Close price must be between low and high") failed += 1 continue history_data = { "stock_id": stock.id, "date": item.date, "open_price": item.open_price, "close_price": item.close_price, "high_price": item.high_price, "low_price": item.low_price, "volume": item.volume, "adjusted_close_price": item.adjusted_close_price, } repository.create_stockhistory(db, history_data) successful += 1 except Exception as e: errors.append(f"Row {idx + 1}: {str(e)}") failed += 1 try: db.commit() except Exception as e: db.rollback() raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Failed to commit bulk import: {str(e)}" ) return BulkImportHistoryResponse( total=total, successful=successful, failed=failed, errors=errors )