Add curated news and bulk position APIs

App/routers/portfolio/routes.py

@@ -47,6 +47,7 @@ from .schemas import (
     BondHoldingCreate,
     BondHoldingUpdate,
     BondSellSchema,
+    BulkPortfolioPositionsUpload,
     CalendarEventCreate,
     CalendarEventResponse,
     TransactionDetailResponse,
@@ -961,6 +962,96 @@ async def add_bond(
     )
 
 
+@router.post("/{portfolio_id}/positions/bulk", summary="Bulk upload portfolio positions")
+async def bulk_upload_positions(
+    portfolio_id: int,
+    payload: BulkPortfolioPositionsUpload,
+    background_tasks: BackgroundTasks,
+    current_user=Depends(get_current_user),
+):
+    await _verify_ownership(portfolio_id, current_user, active_only=True)
+    results = []
+    earliest_date = None
+
+    for index, item in enumerate(payload.positions, start=1):
+        asset_type = item.asset_type.upper()
+        try:
+            if asset_type == "STOCK":
+                stock_id = item.asset_id
+                if not stock_id and item.symbol:
+                    stock = await Stock.get_or_none(symbol=item.symbol.upper())
+                    if not stock:
+                        stock = await Stock.create(symbol=item.symbol.upper(), name=item.name or item.symbol.upper())
+                    stock_id = stock.id
+                if not stock_id or not item.quantity:
+                    raise AppException(status_code=400, message="Stock rows require asset_id or symbol, plus quantity")
+                holding = await PortfolioService.add_stock(
+                    portfolio_id=portfolio_id,
+                    stock_id=stock_id,
+                    quantity=item.quantity,
+                    purchase_price=item.purchase_price,
+                    purchase_date=item.purchase_date,
+                    notes=item.notes,
+                )
+                result = {"index": index, "asset_type": asset_type, "status": "created", "holding_id": holding.id}
+
+            elif asset_type in {"FUND", "UTT"}:
+                fund_id = item.asset_id
+                if not fund_id and item.name:
+                    fund = await MutualFund.get_or_none(name__iexact=item.name)
+                    fund_id = fund.id if fund else None
+                units = item.units_held or item.quantity
+                if not fund_id or not units:
+                    raise AppException(status_code=400, message="Fund rows require asset_id or exact name, plus units_held")
+                holding = await PortfolioService.add_fund(
+                    portfolio_id=portfolio_id,
+                    fund_id=fund_id,
+                    units=units,
+                    purchase_price=item.purchase_price,
+                    purchase_date=item.purchase_date,
+                    notes=item.notes,
+                )
+                result = {"index": index, "asset_type": asset_type, "status": "created", "holding_id": holding.id}
+
+            elif asset_type == "BOND":
+                if not item.asset_id or not item.face_value_held:
+                    raise AppException(status_code=400, message="Bond rows require asset_id and face_value_held")
+                total_price = item.face_value_held * (item.price_per_100 or item.purchase_price) / Decimal("100")
+                holding = await PortfolioService.add_bond(
+                    portfolio_id=portfolio_id,
+                    bond_id=item.asset_id,
+                    face_value=item.face_value_held,
+                    total_purchase_price=total_price,
+                    purchase_date=item.purchase_date,
+                    notes=item.notes,
+                    holding_number=item.holding_number,
+                    holding_status=item.holding_status,
+                )
+                result = {"index": index, "asset_type": asset_type, "status": "created", "holding_id": holding.id}
+
+            else:
+                raise AppException(status_code=400, message=f"Unsupported asset_type {item.asset_type}")
+
+            if earliest_date is None or item.purchase_date < earliest_date:
+                earliest_date = item.purchase_date
+            results.append(result)
+        except Exception as exc:
+            error = getattr(exc, "message", str(exc))
+            results.append({"index": index, "asset_type": asset_type, "status": "failed", "error": error})
+            if not payload.continue_on_error:
+                break
+
+    recalc = None
+    if earliest_date:
+        recalc = await _queue_snapshot_regeneration(portfolio_id, earliest_date, background_tasks)
+
+    return ResponseModel(
+        success=not any(row["status"] == "failed" for row in results),
+        message="Bulk position upload completed",
+        data={"results": results, "recalculation": recalc},
+    )
+
+
 @router.post("/{portfolio_id}/bonds/{bond_id}/sell", summary="Sell bond")
 async def sell_bond(
     portfolio_id: int,

App/routers/portfolio/schemas.py

@@ -192,6 +192,27 @@ class BondSellSchema(BaseModel):
     notes: Optional[str] = None
 
 
+class BulkPortfolioPositionItem(BaseModel):
+    asset_type: str = Field(..., pattern="^(STOCK|FUND|UTT|BOND|stock|fund|utt|bond)$")
+    asset_id: Optional[int] = None
+    symbol: Optional[str] = None
+    name: Optional[str] = None
+    quantity: Optional[Decimal] = Field(None, gt=0)
+    units_held: Optional[Decimal] = Field(None, gt=0)
+    face_value_held: Optional[Decimal] = Field(None, gt=0)
+    purchase_price: Decimal = Field(..., gt=0)
+    price_per_100: Optional[Decimal] = Field(None, gt=0)
+    purchase_date: date
+    notes: Optional[str] = None
+    holding_number: Optional[int] = None
+    holding_status: Optional[str] = None
+
+
+class BulkPortfolioPositionsUpload(BaseModel):
+    positions: list[BulkPortfolioPositionItem] = Field(..., min_length=1, max_length=500)
+    continue_on_error: bool = True
+
+
 class BondHoldingResponse(BaseModel):
     model_config = ConfigDict(from_attributes=True)
 

App/routers/stocks/routes.py

@@ -6,6 +6,8 @@ from .schemas import (
     StockCompanyInfoResponse,
     StockFundamentalSnapshotResponse,
     StockNewsArticleResponse,
+    StockNewsManualCreate,
+    StockNewsCleanupRequest,
     PriceDataResponse,
     StockReferenceSyncPayload,
     CorporateActionResponse,
@@ -24,10 +26,11 @@ from .crud import (
     get_stock_company_info,
     get_cached_stock_news,
     replace_stock_news,
+    upsert_stock_news_article,
 )
 from .service import fetch_dse_stock_data
 from .metrics import calculate_metrics
-from .models import Stock, StockFundamentalSnapshot, StockPriceData, Dividend, StockProfile
+from .models import Stock, StockFundamentalSnapshot, StockPriceData, Dividend, StockProfile, StockNewsArticle
 from .fundamentals import refresh_simplywallst_fundamentals
 from .news_service import (
     build_news_queries,
@@ -379,18 +382,15 @@ async def get_stock_news(
         row
         for row in cached_articles
         if row.section == "industry_news"
-        or is_relevant_company_article(
-            symbol=stock.symbol,
+        or _is_company_news_related(
+            stock=stock,
             stock_name=stock_name,
             headline=row.headline,
             article_text=row.summary or row.article_excerpt or "",
             article_excerpt=row.article_excerpt,
             source_url=row.source_url,
             canonical_url=row.canonical_url,
-            country=country,
-            industry=industry,
-            sector=sector,
-            website=website,
+            company=company,
         )
     ]
     latest_cached_at = cached_articles[0].updated_at if cached_articles else None
@@ -598,6 +598,193 @@ async def get_stock_news(
     )
 
 
+def _clean_manual_news_text(text: str) -> str:
+    replacements = {
+        "¡": "@",
+        "  ": " ",
+    }
+    cleaned = str(text or "").strip()
+    for old, new in replacements.items():
+        cleaned = cleaned.replace(old, new)
+    return " ".join(cleaned.split())
+
+
+def _is_company_news_related(
+    *,
+    stock: Stock,
+    stock_name: str,
+    headline: str,
+    article_text: str,
+    article_excerpt: Optional[str],
+    source_url: Optional[str],
+    canonical_url: Optional[str],
+    company: Optional[dict],
+) -> bool:
+    combined = " ".join(
+        str(part or "")
+        for part in [headline, article_text, article_excerpt, source_url, canonical_url]
+    ).lower()
+    symbol = stock.symbol.upper()
+    if symbol == "TPCC":
+        headline_text = str(headline or "").lower()
+        if "tpcc" in headline_text and not any(token in headline_text for token in ["twiga", "cement", "tanzania portland"]):
+            return False
+        return any(token in combined for token in ["twiga", "tanzania portland", "portland cement", "cement"])
+    if symbol == "TCCL":
+        headline_text = str(headline or "").lower()
+        if "tccl" in headline_text and not any(token in headline_text for token in ["tanga", "simba", "cement"]):
+            return False
+        return any(token in combined for token in ["tanga cement", "simba cement", "cement"])
+    return is_relevant_company_article(
+        symbol=stock.symbol,
+        stock_name=stock_name,
+        headline=headline,
+        article_text=article_text,
+        article_excerpt=article_excerpt,
+        source_url=source_url,
+        canonical_url=canonical_url,
+        country=(company or {}).get("country"),
+        industry=(company or {}).get("industry"),
+        sector=(company or {}).get("sector"),
+        website=(company or {}).get("website"),
+    )
+
+
+def _manual_news_summary(symbol: str, stock_name: str, body: str) -> str:
+    lower = body.lower()
+    if any(word in lower for word in ["mkutano mkuu", "agm", "annual general"]):
+        return (
+            f"{stock_name} has issued an annual general meeting notice. "
+            "Shareholders should review attendance, identification, and proxy deadlines because governance events can affect investor decisions."
+        )
+    return body[:500]
+
+
+@router.post("/{symbol}/news/manual", response_model=ResponseModel)
+async def create_manual_stock_news(
+    symbol: str,
+    payload: StockNewsManualCreate,
+    current_user=Depends(get_current_user),
+):
+    stock = await Stock.get_or_none(symbol=symbol.upper())
+    if not stock:
+        raise AppException(status_code=404, message="Stock not found")
+
+    company = await get_stock_company_info(symbol)
+    stock_name = (company or {}).get("legal_name") or stock.name or stock.symbol
+    body = _clean_manual_news_text(payload.body)
+    headline = _clean_manual_news_text(payload.headline or "")
+    if not headline:
+        headline = f"{stock.symbol}: Shareholder notice"
+
+    relevant = payload.section == "industry_news" or payload.force or _is_company_news_related(
+        stock=stock,
+        stock_name=stock_name,
+        headline=headline,
+        article_text=body,
+        article_excerpt=body,
+        source_url=payload.source_url,
+        canonical_url=payload.source_url,
+        company=company,
+    )
+    if not relevant:
+        raise AppException(
+            status_code=400,
+            message="News does not look related to this company. Use force=true only for verified manual corrections.",
+        )
+
+    if payload.replace_existing:
+        await StockNewsArticle.filter(stock=stock).delete()
+
+    created = await upsert_stock_news_article(
+        stock=stock,
+        query_type=payload.query_type,
+        section=payload.section,
+        headline=headline,
+        source_name=payload.source_name,
+        source_url=payload.source_url,
+        canonical_url=payload.source_url,
+        published_at=payload.published_at or datetime.utcnow(),
+        summary=_manual_news_summary(stock.symbol, stock_name, body),
+        article_excerpt=body,
+        tags=list(dict.fromkeys([*payload.tags, "manual", "company_notice"])),
+        sentiment_label=payload.sentiment_label or "neutral",
+        sentiment_score=payload.sentiment_score if payload.sentiment_score is not None else 0,
+        summary_provider="manual_cleaner",
+        raw_payload={"manual": True, "body": body, "created_by": getattr(current_user, "id", None)},
+    )
+    articles = await get_cached_stock_news(stock, limit=10)
+    data = [
+        StockNewsArticleResponse.model_validate(
+            {
+                "id": row.id,
+                "headline": row.headline,
+                "section": row.section,
+                "query_type": row.query_type,
+                "source_name": row.source_name,
+                "source_url": row.source_url,
+                "canonical_url": row.canonical_url,
+                "published_at": row.published_at,
+                "summary": row.summary,
+                "article_excerpt": row.article_excerpt,
+                "image_url": row.image_url,
+                "extracted_images": row.extracted_images or [],
+                "tags": row.tags or [],
+                "sentiment_label": row.sentiment_label,
+                "sentiment_score": row.sentiment_score,
+                "summary_provider": row.summary_provider,
+            }
+        ).model_dump(mode="json")
+        for row in articles
+    ]
+    return ResponseModel(
+        success=True,
+        message="Manual stock news saved" if created else "Manual stock news updated",
+        data={"articles": data, "created": created},
+    )
+
+
+@router.post("/{symbol}/news/cleanup", response_model=ResponseModel)
+async def cleanup_stock_news(
+    symbol: str,
+    payload: StockNewsCleanupRequest,
+    current_user=Depends(get_current_user),
+):
+    stock = await Stock.get_or_none(symbol=symbol.upper())
+    if not stock:
+        raise AppException(status_code=404, message="Stock not found")
+
+    company = await get_stock_company_info(symbol)
+    stock_name = (company or {}).get("legal_name") or stock.name or stock.symbol
+    rows = await StockNewsArticle.filter(stock=stock).all()
+    remove_ids = set(payload.article_ids)
+    if payload.remove_unrelated:
+        for row in rows:
+            if (row.raw_payload or {}).get("manual"):
+                continue
+            related = row.section == "industry_news" or _is_company_news_related(
+                stock=stock,
+                stock_name=stock_name,
+                headline=row.headline,
+                article_text=row.summary or row.article_excerpt or "",
+                article_excerpt=row.article_excerpt,
+                source_url=row.source_url,
+                canonical_url=row.canonical_url,
+                company=company,
+            )
+            if not related:
+                remove_ids.add(row.id)
+
+    if not payload.dry_run and remove_ids:
+        await StockNewsArticle.filter(stock=stock, id__in=list(remove_ids)).delete()
+
+    return ResponseModel(
+        success=True,
+        message="Stock news cleanup completed",
+        data={"removed_ids": sorted(remove_ids), "dry_run": payload.dry_run},
+    )
+
+
 # --- New Dividend Route ---
 @router.get("/{symbol}/dividends", response_model=ResponseModel)
 async def get_stock_dividends(symbol: str):

App/routers/stocks/schemas.py

@@ -76,6 +76,7 @@ class StockFundamentalSnapshotResponse(BaseModel):
 
 
 class StockNewsArticleResponse(BaseModel):
+    id: Optional[int] = None
     headline: str
     section: str
     query_type: str
@@ -93,6 +94,27 @@ class StockNewsArticleResponse(BaseModel):
     summary_provider: Optional[str] = None
 
 
+class StockNewsManualCreate(BaseModel):
+    headline: Optional[str] = None
+    body: str = Field(..., min_length=5)
+    source_name: Optional[str] = "Manual"
+    source_url: Optional[str] = None
+    published_at: Optional[dt] = None
+    section: str = Field("company_news", pattern="^(company_news|industry_news)$")
+    query_type: str = Field("manual", max_length=30)
+    sentiment_label: Optional[str] = Field(None, pattern="^(positive|negative|neutral)$")
+    sentiment_score: Optional[float] = Field(None, ge=-1, le=1)
+    tags: List[str] = Field(default_factory=list)
+    replace_existing: bool = False
+    force: bool = False
+
+
+class StockNewsCleanupRequest(BaseModel):
+    article_ids: List[int] = Field(default_factory=list)
+    remove_unrelated: bool = True
+    dry_run: bool = False
+
+
 class PriceDataResponse(BaseModel):
     date: date
     opening_price: float