"""SME-Hub investment services — Subprocess SP5. Implements PRD FRSME-INV001 → INV017. Side-effects (notifications, audit log, signal emission) live here, never on the view layer (matches the SP1/SP2/SP3/SP4 convention). Public surface: Investor directory: - register_investor_self, create_investor, update_investor, verify_investor, reject_investor, deactivate_investor, reactivate_investor Funding calls: - create_funding_call, publish_funding_call, close_funding_call, revert_funding_call_to_draft Funding applications: - start_funding_application, auto_save_funding_application, submit_funding_application, decide_funding_application Matchmaking: - match_investors, refresh_matches_for Readiness: - compute_readiness, default_readiness_thresholds Snapshot: - compile_performance_snapshot Investor pitch flow: - request_investor_connection, accept_investor_request, decline_investor_request, withdraw_investor_request Sharing: - share_dashboard_with_investor Manual funding: - record_manual_funding """ from __future__ import annotations import hashlib import json import logging import uuid from collections.abc import Iterable from decimal import Decimal from django.contrib.contenttypes.models import ContentType from django.core.cache import cache from django.core.exceptions import ValidationError from django.db import transaction from django.urls import NoReverseMatch, reverse from django.utils import timezone from apps.core.ai.client import recommend from apps.core.notifications.models import Notification from apps.core.permissions.roles import UserRole from apps.smehub._shared.audit import AuditLog, _audit from apps.smehub._shared.notifications import bulk_notify_smehub, notify_smehub from apps.smehub.investment import signals as inv_signals from apps.smehub.investment.models import ( DashboardAccessLog, FundingApplication, FundingApplicationDocument, FundingCall, InvestmentReadiness, Investor, InvestorConnectionRequest, InvestorMatch, ManualFundingRecord, ReadinessThresholdConfig, SMEPerformanceSnapshot, ) from apps.smehub.onboarding.models import Business, EntrepreneurProfile logger = logging.getLogger(__name__) Verb = Notification.Verb # --------------------------------------------------------------------------- # Exceptions # --------------------------------------------------------------------------- class InvestmentStateError(Exception): """Raised on illegal SP5 transitions / preconditions.""" # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _safe_reverse(name: str, **kwargs) -> str: try: return reverse(name, kwargs=kwargs) if kwargs else reverse(name) except NoReverseMatch: return "" def _admin_users(): from django.contrib.auth import get_user_model User = get_user_model() return User.objects.filter( role__in=[UserRole.SYSTEM_ADMIN, UserRole.ADMIN, UserRole.SME_ADMIN], is_active=True, ) def _eligible_entrepreneur_users(call: FundingCall) -> list: """FRSME-INV005 — entrepreneurs with a verified business that matches the call's target sectors / stages / countries. """ qs = ( Business.objects.filter(verification_status=Business.Status.VERIFIED) .select_related("entrepreneur__user") ) if call.target_sectors: qs = qs.filter(sector__in=list(call.target_sectors)) if call.target_stages: qs = qs.filter(business_stage__in=list(call.target_stages)) if call.target_countries: qs = qs.filter(country__in=list(call.target_countries)) # FRSME-INV004 — structured eligibility criteria. if call.min_full_time_staff: qs = qs.filter(baseline__fte_count__gte=call.min_full_time_staff) if call.require_incubation_complete: from apps.smehub.incubation.models import CohortMember completed_business_ids = CohortMember.objects.filter( status=CohortMember.Status.PROGRAMME_COMPLETE, ).values_list("business_id", flat=True) qs = qs.filter(pk__in=completed_business_ids) user_ids = qs.values_list("entrepreneur__user_id", flat=True).distinct() from django.contrib.auth import get_user_model User = get_user_model() return list(User.objects.filter(pk__in=user_ids, is_active=True)) # --------------------------------------------------------------------------- # Investor directory (FRSME-INV001–003) # --------------------------------------------------------------------------- @transaction.atomic def create_investor( *, by_user, org_name: str, type: str, investment_focus: Iterable[str] | None = None, geographic_preference: Iterable[str] | None = None, ticket_size_min: Decimal | None = None, ticket_size_max: Decimal | None = None, contact_email: str = "", contact_phone: str = "", website: str = "", description: str = "", logo=None, contact_user=None, auto_verify: bool = False, ) -> Investor: investor = Investor.objects.create( org_name=org_name, type=type, investment_focus=list(investment_focus or []), geographic_preference=list(geographic_preference or []), ticket_size_min=ticket_size_min, ticket_size_max=ticket_size_max, contact_email=contact_email, contact_phone=contact_phone, website=website, description=description, logo=logo, contact_user=contact_user, ) _audit(by_user, AuditLog.Action.CREATE, investor) inv_signals.smehub_investor_registered.send( sender=Investor, investor=investor, self_registered=False, by_user=by_user, ) if auto_verify: verify_investor(investor, by_user=by_user) investor = Investor.objects.get(pk=investor.pk) return investor @transaction.atomic def register_investor_self(*, by_user, **fields) -> Investor: """FRSME-INV002 — investor account self-registers; lands in PENDING.""" investor = create_investor( by_user=by_user, contact_user=by_user, **fields, ) inv_signals.smehub_investor_registered.send( sender=Investor, investor=investor, self_registered=True, by_user=by_user, ) bulk_notify_smehub( list(_admin_users()), f"New investor self-registered: {investor.org_name}.", verb=Verb.SMEHUB_INVESTOR_REQUEST, action_url=_safe_reverse("smehub_investment:investor_admin_verify", pk=investor.pk), ) return investor @transaction.atomic def update_investor(investor: Investor, *, by_user, **fields) -> Investor: json_fields = {"investment_focus", "geographic_preference"} for key, value in fields.items(): if key in json_fields and value is not None: setattr(investor, key, list(value)) elif value is not None: setattr(investor, key, value) investor.save() _audit(by_user, AuditLog.Action.UPDATE, investor, metadata={"fields": list(fields.keys())}) return investor @transaction.atomic def verify_investor(investor: Investor, *, by_user) -> Investor: investor.verify(by_user=by_user) investor.save() _audit(by_user, AuditLog.Action.UPDATE, investor, metadata={"status": "verified"}) inv_signals.smehub_investor_verified.send(sender=Investor, investor=investor, by_user=by_user) if investor.contact_user is not None: notify_smehub( investor.contact_user, f"Your investor profile “{investor.org_name}” has been verified.", verb=Verb.SMEHUB_INVESTOR_ACCEPTED, target=investor, action_url=_safe_reverse("smehub_investment:investor_detail", pk=investor.pk), ) return investor @transaction.atomic def reject_investor(investor: Investor, *, by_user, reason: str) -> Investor: investor.reject(reason=reason) investor.save() _audit(by_user, AuditLog.Action.UPDATE, investor, metadata={"status": "rejected", "reason": reason}) inv_signals.smehub_investor_rejected.send( sender=Investor, investor=investor, by_user=by_user, reason=reason, ) if investor.contact_user is not None: notify_smehub( investor.contact_user, f"Your investor profile “{investor.org_name}” was rejected.", verb=Verb.SMEHUB_INVESTOR_DECLINED, target=investor, ) return investor @transaction.atomic def deactivate_investor(investor: Investor, *, by_user, reason: str = "") -> Investor: investor.deactivate(reason=reason) investor.save() _audit(by_user, AuditLog.Action.UPDATE, investor, metadata={"status": "deactivated"}) inv_signals.smehub_investor_deactivated.send(sender=Investor, investor=investor, by_user=by_user) return investor @transaction.atomic def reactivate_investor(investor: Investor, *, by_user) -> Investor: investor.reactivate(by_user=by_user) investor.save() _audit(by_user, AuditLog.Action.UPDATE, investor, metadata={"status": "verified"}) inv_signals.smehub_investor_verified.send(sender=Investor, investor=investor, by_user=by_user) return investor # --------------------------------------------------------------------------- # Funding calls (FRSME-INV004–005) # --------------------------------------------------------------------------- @transaction.atomic def create_funding_call( *, by_user, title: str, type: str, application_deadline, summary: str = "", description: str = "", eligibility: dict | None = None, target_sectors: Iterable[str] | None = None, target_stages: Iterable[str] | None = None, target_countries: Iterable[str] | None = None, amount_min: Decimal | None = None, amount_max: Decimal | None = None, contact_email: str = "", cover_image=None, investor: Investor | None = None, min_full_time_staff: int | None = None, require_incubation_complete: bool = False, ) -> FundingCall: call = FundingCall.objects.create( title=title, summary=summary, description=description, type=type, application_deadline=application_deadline, eligibility=eligibility or {}, target_sectors=list(target_sectors or []), target_stages=list(target_stages or []), target_countries=list(target_countries or []), amount_min=amount_min, amount_max=amount_max, contact_email=contact_email, cover_image=cover_image, organiser=by_user, investor=investor, min_full_time_staff=min_full_time_staff, require_incubation_complete=require_incubation_complete, ) _audit(by_user, AuditLog.Action.CREATE, call) return call @transaction.atomic def publish_funding_call(call: FundingCall, *, by_user) -> FundingCall: """FRSME-INV004 + 005 — publish + bulk-notify eligible entrepreneurs.""" if call.status != FundingCall.Status.DRAFT: raise InvestmentStateError("Only draft funding calls can be published.") call.publish() call.save() _audit(by_user, AuditLog.Action.UPDATE, call, metadata={"status": "published"}) inv_signals.smehub_funding_call_published.send( sender=FundingCall, funding_call=call, by_user=by_user, ) bulk_notify_smehub( _eligible_entrepreneur_users(call), f"New funding call published: “{call.title}”. Application deadline {call.application_deadline:%d %b %Y}.", verb=Verb.SMEHUB_FUNDING_CALL_PUBLISHED, action_url=_safe_reverse("smehub_investment:funding_call_detail", pk=call.pk), ) return call @transaction.atomic def close_funding_call(call: FundingCall, *, by_user) -> FundingCall: if call.status != FundingCall.Status.PUBLISHED: raise InvestmentStateError("Only published funding calls can be closed.") call.close() call.save() _audit(by_user, AuditLog.Action.UPDATE, call, metadata={"status": "closed"}) inv_signals.smehub_funding_call_closed.send( sender=FundingCall, funding_call=call, by_user=by_user, ) return call @transaction.atomic def revert_funding_call_to_draft(call: FundingCall, *, by_user) -> FundingCall: if call.status != FundingCall.Status.PUBLISHED: raise InvestmentStateError("Only published funding calls can be reverted.") call.revert_to_draft() call.save() _audit(by_user, AuditLog.Action.UPDATE, call, metadata={"status": "draft"}) return call # --------------------------------------------------------------------------- # Funding applications (FRSME-INV006–007) # --------------------------------------------------------------------------- def _check_business(business: Business, entrepreneur: EntrepreneurProfile) -> None: if business.entrepreneur_id != entrepreneur.pk: raise InvestmentStateError("Business does not belong to this entrepreneur.") if business.verification_status != Business.Status.VERIFIED: raise InvestmentStateError("Business must be verified before applying.") @transaction.atomic def start_funding_application( *, funding_call: FundingCall, entrepreneur: EntrepreneurProfile, business: Business, ) -> FundingApplication: """Create a DRAFT row for HTMX auto-save. Idempotent on (call, business).""" if funding_call.status != FundingCall.Status.PUBLISHED: raise InvestmentStateError("Funding call is not accepting applications.") if funding_call.application_deadline and timezone.now() > funding_call.application_deadline: raise InvestmentStateError("Application deadline has passed.") _check_business(business, entrepreneur) existing = FundingApplication.objects.filter( funding_call=funding_call, business=business, status__in=[ FundingApplication.Status.DRAFT, FundingApplication.Status.SUBMITTED, FundingApplication.Status.UNDER_REVIEW, FundingApplication.Status.RETURNED_FOR_INFO, ], ).first() if existing: return existing application = FundingApplication.objects.create( funding_call=funding_call, entrepreneur=entrepreneur, business=business, ) _audit(entrepreneur.user, AuditLog.Action.CREATE, application) return application @transaction.atomic def auto_save_funding_application( application: FundingApplication, *, payload: dict, ) -> FundingApplication: """HTMX auto-save buffer (mirror SP2 incubation auto-save).""" if application.status != FundingApplication.Status.DRAFT: return application application.auto_saved_payload = dict(payload or {}) application.save(update_fields=["auto_saved_payload", "updated_at"]) return application @transaction.atomic def submit_funding_application( application: FundingApplication, *, by_user, ) -> FundingApplication: if application.status not in ( FundingApplication.Status.DRAFT, FundingApplication.Status.RETURNED_FOR_INFO, ): raise InvestmentStateError("Application can only be submitted from draft / returned states.") if application.status == FundingApplication.Status.DRAFT: application.submit() else: application.resubmit() application.save() _audit(by_user, AuditLog.Action.UPDATE, application, metadata={"status": "submitted"}) inv_signals.smehub_funding_application_submitted.send( sender=FundingApplication, application=application, ) if application.funding_call.organiser is not None: notify_smehub( application.funding_call.organiser, f"New funding application from {application.business.name} for “{application.funding_call.title}”.", verb=Verb.SMEHUB_APPLICATION_RECEIVED, action_url=_safe_reverse( "smehub_investment:funding_call_applications", pk=application.funding_call.pk, ), target=application, ) return application @transaction.atomic def decide_funding_application( application: FundingApplication, *, action: str, by_user, note: str = "", ) -> FundingApplication: """FRSME-INV007 — accept / reject / return-for-info on a submitted app.""" if action == "accept": application.accept(by_user=by_user, note=note) elif action == "reject": application.reject(by_user=by_user, note=note) elif action == "return_for_info": application.return_for_info(message=note) else: raise InvestmentStateError(f"Unknown decision action: {action}") application.save() _audit(by_user, AuditLog.Action.UPDATE, application, metadata={"action": action}) inv_signals.smehub_funding_application_decided.send( sender=FundingApplication, application=application, action=action, by_user=by_user, ) notify_smehub( application.entrepreneur.user, f"Your funding application for “{application.funding_call.title}” was {action.replace('_', ' ')}.", verb=Verb.SMEHUB_FUNDING_OUTCOME, target=application, action_url=_safe_reverse( "smehub_investment:funding_application_detail", pk=application.pk, ), ) if action == "accept": amount = application.funding_request_amount or Decimal("0") inv_signals.smehub_funding_secured.send( sender=FundingApplication, entrepreneur=application.entrepreneur, business=application.business, amount=amount, source="application", ) notify_smehub( application.entrepreneur.user, f"Funding secured: “{application.funding_call.title}” (USD {amount}).", verb=Verb.SMEHUB_FUNDING_SECURED, target=application, ) return application # --------------------------------------------------------------------------- # Investor matchmaking (FRSME-INV008–009) # --------------------------------------------------------------------------- def _entrepreneur_features(entrepreneur: EntrepreneurProfile) -> dict: businesses = list( Business.objects.filter( entrepreneur=entrepreneur, verification_status=Business.Status.VERIFIED, ).values("sector", "business_stage", "country"), ) sectors = sorted({b["sector"] for b in businesses if b["sector"]}) stages = sorted({b["business_stage"] for b in businesses if b["business_stage"]}) geo = sorted({b["country"] for b in businesses if b["country"]}) return { "sectors": sectors, "stages": stages, "geo": geo, "stage": stages[0] if stages else "", } def _investor_payload(qs: Iterable[Investor]) -> list[dict]: payload: list[dict] = [] for inv in qs: payload.append( { "id": inv.pk, "label": inv.org_name, "kind": "investor", "type": inv.type, "sectors": list(inv.investment_focus or []), "geo": list(inv.geographic_preference or []), # FRSME-INV008 audit fix: actually use the investor's # ``investment_stages`` instead of feeding the matchmaker an # empty list (which silently zeroed out the 15% stage weight). "stages": list(getattr(inv, "investment_stages", []) or []), "ticket_min": float(inv.ticket_size_min or 0), "ticket_max": float(inv.ticket_size_max or 0), } ) return payload def match_investors(entrepreneur: EntrepreneurProfile, *, top_k: int = 12) -> list[dict]: """Return ranked investor recommendations without persisting.""" profile = _entrepreneur_features(entrepreneur) investors = Investor.objects.filter(verification_status=Investor.Status.VERIFIED) candidates = _investor_payload(investors) return recommend(profile, candidates, top_k=top_k) @transaction.atomic def refresh_matches_for( entrepreneur: EntrepreneurProfile, *, trigger: str = "manual", top_k: int = 12, ) -> int: """Compute + persist InvestorMatch rows for an entrepreneur. Idempotent.""" profile = _entrepreneur_features(entrepreneur) investors = list(Investor.objects.filter(verification_status=Investor.Status.VERIFIED)) candidates = _investor_payload(investors) ranked = recommend(profile, candidates, top_k=top_k) is_fallback = bool(ranked and ranked[0].get("is_fallback", True)) seen_ids: list[int] = [] for entry in ranked: investor_id = entry.get("id") if not investor_id: continue InvestorMatch.objects.update_or_create( entrepreneur=entrepreneur, investor_id=investor_id, defaults={ "score": Decimal(str(entry.get("score", 0))), "reason": entry.get("reason", ""), "is_fallback": entry.get("is_fallback", is_fallback), "generated_at": timezone.now(), "refreshed_by_signal": trigger, }, ) seen_ids.append(investor_id) # Drop matches for investors no longer in the verified pool. InvestorMatch.objects.filter(entrepreneur=entrepreneur).exclude( investor_id__in=seen_ids, ).delete() inv_signals.smehub_investor_match_refreshed.send( sender=InvestorMatch, entrepreneur=entrepreneur, trigger=trigger, count=len(seen_ids), ) return len(seen_ids) # --------------------------------------------------------------------------- # Investment readiness (FRSME-INV010) # --------------------------------------------------------------------------- _HARDCODED_READINESS_THRESHOLDS = { "verified_business_required": True, "baseline_required": True, "incubation_completion_required": False, "min_partnerships": 1, "min_showcase_or_catalogue": 0, "min_funding_events": 0, } def default_readiness_thresholds() -> dict: """FRSME-INV010 — the active admin-configured thresholds, falling back to the built-in defaults when none is configured. Per-call overrides still layer on top via ``compute_readiness(thresholds=...)``. """ cfg = ( ReadinessThresholdConfig.objects.filter(is_active=True) .order_by("-updated_at") .first() ) if cfg is not None: return cfg.as_thresholds() return dict(_HARDCODED_READINESS_THRESHOLDS) def _readiness_actions(breakdown: dict) -> list[dict]: actions: list[dict] = [] if not breakdown.get("verified_business"): actions.append({ "action": "Register and verify a business", "reason": "Investors require a verified business profile.", "link": _safe_reverse("smehub_onboarding:business_register"), }) if not breakdown.get("baseline_set"): actions.append({ "action": "Complete your business baseline", "reason": "Baseline data anchors the M&E and investor view.", "link": "", }) if not breakdown.get("has_partnerships"): actions.append({ "action": "Form at least one partnership", "reason": "Partnerships demonstrate validation and capability.", "link": _safe_reverse("smehub_linkage:partner_directory"), }) if not breakdown.get("has_commercialisation"): actions.append({ "action": "Showcase your innovation or publish to the catalogue", "reason": "Visible commercialisation signals traction.", "link": _safe_reverse("smehub_showcasing:event_list"), }) return actions @transaction.atomic def compute_readiness( entrepreneur: EntrepreneurProfile, *, thresholds: dict | None = None, ) -> InvestmentReadiness: """Evaluate readiness flag against thresholds and persist.""" cfg = {**default_readiness_thresholds(), **(thresholds or {})} snapshot = _read_snapshot(entrepreneur) payload = snapshot.payload if snapshot else {} sp1 = payload.get("sp1", {}) or {} sp2 = payload.get("sp2", {}) or {} sp3 = payload.get("sp3", {}) or {} sp4 = payload.get("sp4", {}) or {} sp5 = payload.get("sp5", {}) or {} breakdown = { "verified_business": bool(sp1.get("verified_businesses", 0)), "baseline_set": bool(sp1.get("has_baseline")), "incubation_complete": bool(sp2.get("programmes_completed", 0)), "has_partnerships": (sp3.get("active_connections", 0) or 0) >= cfg["min_partnerships"], "has_commercialisation": ( (sp4.get("catalogue_entries", 0) or 0) + (sp4.get("confirmed_showcases", 0) or 0) ) >= cfg["min_showcase_or_catalogue"], "has_funding_events": (sp5.get("funding_events", 0) or 0) >= cfg["min_funding_events"], } is_ready = ( (breakdown["verified_business"] or not cfg["verified_business_required"]) and (breakdown["baseline_set"] or not cfg["baseline_required"]) and (breakdown["incubation_complete"] or not cfg["incubation_completion_required"]) and breakdown["has_partnerships"] and breakdown["has_commercialisation"] and breakdown["has_funding_events"] ) record, _ = InvestmentReadiness.objects.update_or_create( entrepreneur=entrepreneur, defaults={ "is_ready": is_ready, "threshold_breakdown": breakdown, "recommended_actions": _readiness_actions(breakdown), "last_evaluated_at": timezone.now(), }, ) inv_signals.smehub_investment_readiness_evaluated.send( sender=InvestmentReadiness, entrepreneur=entrepreneur, is_ready=is_ready, ) return record # --------------------------------------------------------------------------- # Performance snapshot (FRSME-INV011–012) # --------------------------------------------------------------------------- SNAPSHOT_CACHE_PREFIX = "smehub:performance_snapshot:" SNAPSHOT_CACHE_TTL = 60 * 60 # 1 hour def _read_snapshot(entrepreneur: EntrepreneurProfile) -> SMEPerformanceSnapshot | None: return SMEPerformanceSnapshot.objects.filter(entrepreneur=entrepreneur).first() def _baseline_payload(entrepreneur: EntrepreneurProfile) -> dict: businesses = list( Business.objects.filter(entrepreneur=entrepreneur).values( "id", "name", "sector", "business_stage", "country", "verification_status", ) ) verified = [b for b in businesses if b["verification_status"] == Business.Status.VERIFIED] has_baseline = False try: from apps.smehub.onboarding.models import BusinessBaseline has_baseline = BusinessBaseline.objects.filter( business__entrepreneur=entrepreneur, ).exists() except Exception: pass return { "businesses_total": len(businesses), "verified_businesses": len(verified), "primary_business": verified[0] if verified else (businesses[0] if businesses else None), "has_baseline": has_baseline, "sectors": sorted({b["sector"] for b in businesses if b["sector"]}), "stages": sorted({b["business_stage"] for b in businesses if b["business_stage"]}), "countries": sorted({b["country"] for b in businesses if b["country"]}), "entrepreneur_verified": ( entrepreneur.verification_status == EntrepreneurProfile.Status.VERIFIED ), } def _incubation_payload(entrepreneur: EntrepreneurProfile) -> dict: payload = { "applications_total": 0, "applications_accepted": 0, "programmes_completed": 0, "active_cohorts": 0, "certificates_issued": 0, } try: from apps.smehub.incubation.models import ( Application, Certificate, CohortMember, ) except Exception: return payload apps_qs = Application.objects.filter(entrepreneur=entrepreneur) payload["applications_total"] = apps_qs.count() payload["applications_accepted"] = apps_qs.filter(status=Application.Status.ACCEPTED).count() members = CohortMember.objects.filter(entrepreneur=entrepreneur) payload["active_cohorts"] = members.filter(status=CohortMember.Status.ACTIVE).count() payload["programmes_completed"] = members.filter( status=CohortMember.Status.PROGRAMME_COMPLETE, ).count() payload["certificates_issued"] = Certificate.objects.filter( cohort_member__entrepreneur=entrepreneur, ).count() return payload def _linkage_payload(entrepreneur: EntrepreneurProfile) -> dict: payload = { "active_connections": 0, "formalised_mous": 0, "advisory_completed": 0, } try: from apps.smehub.linkage.models import AdvisorySession, Connection, MoU except Exception: return payload payload["active_connections"] = Connection.objects.filter( entrepreneur=entrepreneur, status=Connection.Status.ACTIVE, ).count() payload["formalised_mous"] = MoU.objects.filter( connection__entrepreneur=entrepreneur, status__in=[MoU.Status.FORMALISED, MoU.Status.AMENDED], ).count() payload["advisory_completed"] = AdvisorySession.objects.filter( entrepreneur=entrepreneur, status=AdvisorySession.Status.COMPLETED, ).count() return payload def _commercialisation_payload(entrepreneur: EntrepreneurProfile) -> dict: payload = { "showcase_applications": 0, "confirmed_showcases": 0, "catalogue_entries": 0, "deal_rooms_open": 0, "agreements_formalised": 0, } try: from apps.smehub.showcasing.models import ( DealAgreement, DealRoom, InnovationCatalogueEntry, ShowcaseApplication, ) except Exception: return payload payload["showcase_applications"] = ShowcaseApplication.objects.filter( entrepreneur=entrepreneur, ).count() payload["confirmed_showcases"] = ShowcaseApplication.objects.filter( entrepreneur=entrepreneur, status=ShowcaseApplication.Status.CONFIRMED, ).count() payload["catalogue_entries"] = InnovationCatalogueEntry.objects.filter( entrepreneur=entrepreneur, is_active=True, ).count() payload["deal_rooms_open"] = DealRoom.objects.filter( entrepreneur=entrepreneur, status=DealRoom.Status.OPEN, ).count() payload["agreements_formalised"] = DealAgreement.objects.filter( deal_room__entrepreneur=entrepreneur, ).count() return payload def _investment_payload(entrepreneur: EntrepreneurProfile) -> dict: apps_qs = FundingApplication.objects.filter(entrepreneur=entrepreneur) accepted = apps_qs.filter(status=FundingApplication.Status.ACCEPTED) accepted_total = sum( (a.funding_request_amount or Decimal("0")) for a in accepted ) manual_qs = ManualFundingRecord.objects.filter(entrepreneur=entrepreneur) manual_total = sum((r.amount for r in manual_qs), Decimal("0")) return { "funding_applications": apps_qs.count(), "funding_accepted": accepted.count(), "funding_accepted_amount": float(accepted_total), "manual_funding_events": manual_qs.count(), "manual_funding_amount": float(manual_total), "funding_events": accepted.count() + manual_qs.count(), "total_funding_secured": float(accepted_total + manual_total), "investor_requests_pending": InvestorConnectionRequest.objects.filter( entrepreneur=entrepreneur, status=InvestorConnectionRequest.Status.PENDING, ).count(), "investor_requests_accepted": InvestorConnectionRequest.objects.filter( entrepreneur=entrepreneur, status=InvestorConnectionRequest.Status.ACCEPTED, ).count(), } @transaction.atomic def compile_performance_snapshot( entrepreneur: EntrepreneurProfile, *, trigger: str = "manual", ) -> SMEPerformanceSnapshot: """FRSME-INV011–012 — refresh the cross-SP snapshot.""" sp1 = _baseline_payload(entrepreneur) sp2 = _incubation_payload(entrepreneur) sp3 = _linkage_payload(entrepreneur) sp4 = _commercialisation_payload(entrepreneur) sp5 = _investment_payload(entrepreneur) primary_business_id = (sp1.get("primary_business") or {}).get("id") primary_business = None if primary_business_id: primary_business = Business.objects.filter(pk=primary_business_id).first() payload = { "compiled_at": timezone.now().isoformat(), "sp1": sp1, "sp2": sp2, "sp3": sp3, "sp4": sp4, "sp5": sp5, } snapshot, _ = SMEPerformanceSnapshot.objects.update_or_create( entrepreneur=entrepreneur, defaults={ "business": primary_business, "payload": payload, "version": ( SMEPerformanceSnapshot.objects.filter(entrepreneur=entrepreneur) .values_list("version", flat=True) .first() or 0 ) + 1, "refreshed_at": timezone.now(), "refreshed_by_signal": trigger, }, ) cache.set( f"{SNAPSHOT_CACHE_PREFIX}{entrepreneur.pk}", payload, timeout=SNAPSHOT_CACHE_TTL, ) inv_signals.smehub_performance_snapshot_refreshed.send( sender=SMEPerformanceSnapshot, snapshot=snapshot, trigger=trigger, ) return snapshot # --------------------------------------------------------------------------- # Investor connection requests (FRSME-INV013–015) # --------------------------------------------------------------------------- @transaction.atomic def request_investor_connection( *, entrepreneur: EntrepreneurProfile, business: Business, investor: Investor, message: str = "", funding_ask=None, supporting_link: str = "", ) -> InvestorConnectionRequest: _check_business(business, entrepreneur) if investor.verification_status != Investor.Status.VERIFIED: raise InvestmentStateError("Investor must be verified before sending a pitch.") existing = InvestorConnectionRequest.objects.filter( entrepreneur=entrepreneur, business=business, investor=investor, status__in=[ InvestorConnectionRequest.Status.PENDING, InvestorConnectionRequest.Status.ACCEPTED, ], ).first() if existing: return existing request = InvestorConnectionRequest.objects.create( entrepreneur=entrepreneur, business=business, investor=investor, message=message, funding_ask=funding_ask, supporting_link=supporting_link, ) _audit(entrepreneur.user, AuditLog.Action.CREATE, request) inv_signals.smehub_investor_request_submitted.send( sender=InvestorConnectionRequest, request=request, ) if investor.contact_user is not None: notify_smehub( investor.contact_user, f"{business.name} sent you a pitch request.", verb=Verb.SMEHUB_INVESTOR_REQUEST, target=request, action_url=_safe_reverse("smehub_investment:investor_request_detail", pk=request.pk), ) else: bulk_notify_smehub( list(_admin_users()), f"{business.name} requested a pitch with {investor.org_name}.", verb=Verb.SMEHUB_INVESTOR_REQUEST, action_url=_safe_reverse("smehub_investment:investor_request_detail", pk=request.pk), ) return request @transaction.atomic def accept_investor_request( request: InvestorConnectionRequest, *, by_user, ) -> InvestorConnectionRequest: """FRSME-INV015 — opening a DealRoom delegates to SP4 so deal-room infra remains the single source of truth (memory note). """ if request.status != InvestorConnectionRequest.Status.PENDING: raise InvestmentStateError("Only pending requests can be accepted.") from apps.smehub.showcasing.services import open_deal_room request.accept(by_user=by_user) request.save() deal_room = open_deal_room( entrepreneur=request.entrepreneur, business=request.business, counterparty=request.investor, origin=request, title=f"{request.business.name} ↔ {request.investor.org_name}", summary=request.message[:1000] if request.message else "", created_by=by_user, ) InvestorConnectionRequest.objects.filter(pk=request.pk).update(deal_room=deal_room) request = InvestorConnectionRequest.objects.get(pk=request.pk) _audit(by_user, AuditLog.Action.UPDATE, request, metadata={"status": "accepted"}) inv_signals.smehub_investor_request_accepted.send( sender=InvestorConnectionRequest, request=request, deal_room=deal_room, by_user=by_user, ) notify_smehub( request.entrepreneur.user, f"{request.investor.org_name} accepted your pitch — a deal room is open.", verb=Verb.SMEHUB_INVESTOR_ACCEPTED, target=request, action_url=_safe_reverse("smehub_showcasing:deal_room_detail", pk=deal_room.pk), ) # FRSME-INV012 audit fix: PRD specifies the SME Performance Dashboard # is "made accessible to a matched investor during due diligence review, # *upon acceptance* of a connection request" (Table 43). Audit found # this auto-share was missing — the entrepreneur had to share manually. # Best-effort: if sharing fails (e.g. investor lost VERIFIED status), # log it but don't block acceptance. try: share_dashboard_with_investor( entrepreneur=request.entrepreneur, investor=request.investor, scope=DashboardAccessLog.Scope.SUMMARY, note=f"Auto-shared on pitch acceptance (request #{request.pk}).", ) except InvestmentStateError: logger.warning( "Auto-share of dashboard skipped on accept of request %s — investor not verified.", request.pk, ) return request @transaction.atomic def decline_investor_request( request: InvestorConnectionRequest, *, by_user, reason: str = "", ) -> InvestorConnectionRequest: if request.status != InvestorConnectionRequest.Status.PENDING: raise InvestmentStateError("Only pending requests can be declined.") request.decline(by_user=by_user, reason=reason) request.save() _audit(by_user, AuditLog.Action.UPDATE, request, metadata={"status": "declined"}) inv_signals.smehub_investor_request_declined.send( sender=InvestorConnectionRequest, request=request, by_user=by_user, reason=reason, ) notify_smehub( request.entrepreneur.user, f"{request.investor.org_name} declined your pitch request.", verb=Verb.SMEHUB_INVESTOR_DECLINED, target=request, ) return request @transaction.atomic def withdraw_investor_request( request: InvestorConnectionRequest, *, by_user, ) -> InvestorConnectionRequest: if request.status != InvestorConnectionRequest.Status.PENDING: raise InvestmentStateError("Only pending requests can be withdrawn.") request.withdraw() request.save() _audit(by_user, AuditLog.Action.UPDATE, request, metadata={"status": "withdrawn"}) return request # --------------------------------------------------------------------------- # Dashboard sharing (FRSME-INV012) # --------------------------------------------------------------------------- def _dashboard_share_ttl(): """FRSME-INV012 audit fix — bounded share lifetime.""" from datetime import timedelta from django.conf import settings as dj_settings days = int(getattr(dj_settings, "SMEHUB_DASHBOARD_SHARE_TTL_DAYS", 90)) return timedelta(days=max(days, 1)) @transaction.atomic def share_dashboard_with_investor( *, entrepreneur: EntrepreneurProfile, investor: Investor, scope: str = DashboardAccessLog.Scope.SUMMARY, note: str = "", ) -> DashboardAccessLog: """Create a DashboardAccessLog row + return it (the share token in the row is what gets embedded in the signed URL). The share expires after ``SMEHUB_DASHBOARD_SHARE_TTL_DAYS`` (default 90 days) so a leaked URL has bounded blast radius. """ if investor.verification_status != Investor.Status.VERIFIED: raise InvestmentStateError("Investor must be verified before sharing.") log = DashboardAccessLog.objects.create( entrepreneur=entrepreneur, investor=investor, viewer_content_type=ContentType.objects.get_for_model(Investor), viewer_object_id=str(investor.pk), scope=scope, note=note, expires_at=timezone.now() + _dashboard_share_ttl(), ) _audit(entrepreneur.user, AuditLog.Action.ACCESS, log, metadata={"scope": scope}) inv_signals.smehub_dashboard_shared.send(sender=DashboardAccessLog, log=log) if investor.contact_user is not None: notify_smehub( investor.contact_user, f"{entrepreneur.user.get_full_name() or entrepreneur.user.email} shared a performance dashboard.", verb=Verb.SMEHUB_INVESTOR_REQUEST, target=log, action_url=_safe_reverse( "smehub_investment:performance_dashboard_shared", token=str(log.share_token), ), ) return log # --------------------------------------------------------------------------- # Manual funding records (FRSME-INV016) # --------------------------------------------------------------------------- @transaction.atomic def record_manual_funding( *, by_admin, entrepreneur: EntrepreneurProfile, business: Business, funder_name: str, amount: Decimal, funded_at, currency: str = "USD", investor: Investor | None = None, source_note: str = "", ) -> ManualFundingRecord: if amount is None or Decimal(amount) <= 0: raise ValidationError("Manual funding amount must be positive.") if business.entrepreneur_id != entrepreneur.pk: raise InvestmentStateError("Business does not belong to this entrepreneur.") # FRSME-INV016 — manual funding events must be recorded against a verified # business so the MEI feed (INV017) is not polluted with unvetted records. if business.verification_status != Business.Status.VERIFIED: raise InvestmentStateError( "Cannot record manual funding: the business is not currently verified.", ) record = ManualFundingRecord.objects.create( entrepreneur=entrepreneur, business=business, funder_name=funder_name, investor=investor, amount=Decimal(amount), currency=currency, funded_at=funded_at, source_note=source_note, captured_by_admin=by_admin, ) _audit( by_admin, AuditLog.Action.CREATE, record, metadata={ "funder_name": funder_name, "amount": str(amount), "currency": currency, }, ) inv_signals.smehub_manual_funding_recorded.send( sender=ManualFundingRecord, record=record, by_user=by_admin, ) inv_signals.smehub_funding_secured.send( sender=ManualFundingRecord, entrepreneur=entrepreneur, business=business, amount=Decimal(amount), source="manual", ) notify_smehub( entrepreneur.user, f"Funding event recorded: {funder_name} — {currency} {amount}.", verb=Verb.SMEHUB_FUNDING_SECURED, target=record, ) return record