Robo-Advisory: How to Build It

from datetime import date, timedelta

# STEP 4. TWO DIFFERENT THINGS, ROUTINELY COLLAPSED INTO ONE SCORE.
#
#   CAPACITY  = how much loss this person can absorb without the plan failing.
#               Arithmetic. Horizon, income stability, dependants, emergency
#               fund, the size of this pot relative to everything else.
#   TOLERANCE = how much loss this person can live through without selling.
#               Psychology. Measured by questionnaire, and questionnaires are
#               optimistic on a day the market is calm.
#
# A single "risk score" that blends them is unusable, because the two failure
# modes are opposite: too much capacity wastes return, too much tolerance
# produces a complaint after a drawdown.

def risk_inputs(client):
    capacity = score_capacity(client)          # 1..7
    tolerance = score_tolerance(client)        # 1..7
    required = required_for_goal(client)       # what the GOAL demands
    binding = min(capacity, tolerance)         # never 'required'
    return {
        "capacity": capacity, "tolerance": tolerance,
        "required": required, "binding": binding,
        # The asymmetry that predicts complaints. Flag it, and record that
        # the conversation happened.
        "flag_review": tolerance - capacity >= 2,
        "goal_unreachable": required > binding,
    }

def resolve_goal(inp, client):
    """If the goal needs more risk than the client can bear, the GOAL changes
    -- horizon, contribution or target -- not the portfolio. This is exactly
    the constraint an optimiser will quietly relax if you let it."""
    if not inp["goal_unreachable"]:
        return {"action": "proceed"}
    return {"action": "renegotiate_goal",
            "options": ["extend_horizon", "raise_contribution", "lower_target"],
            "not_an_option": "raise_risk_above_binding"}

# STEP 5. THE RECOMMENDATION, AND THE PART EVERYONE DROPS.
def recommend(universe, inp, policy):
    included, excluded = [], []
    for f in universe:
        why = exclusion_reason(f, inp, policy)   # ONE reason, the first that bites
        (excluded if why else included).append({"fund": f["id"], "reason": why}
                                               if why else {"fund": f["id"]})
    return {
        "as_of": date.today().isoformat(),
        "binding_constraint": inp["binding"],
        "universe_size": len(universe),
        "included": included,
        # "Why was this fund never recommended to this client?" has a correct
        # answer and it CANNOT be recomputed later, because the catalogue, the
        # policy and the client's profile will all have moved.
        "excluded": excluded,
        "policy_version": policy["version"],
        "universe_snapshot_hash": sha256_of(universe),
    }

# WHAT TO CHECK
# [ ] capacity and tolerance stored SEPARATELY, and both shown to the client
# [ ] 'required' is never a permission. If the goal needs more risk than the
#     client can bear, the goal changes
# [ ] tolerance exceeding capacity by two bands or more is flagged, the
#     conversation is had, and the fact that it was had is recorded
# [ ] store EXCLUSIONS, not just the recommendation. Cheap to write, and the
#     difference between inspection-ready and inspection-panicked
# [ ] store the universe as it stood, not a pointer to a catalogue that
#     changes weekly
# [ ] a risk profile has an expiry. So does a recommendation. They are not
#     the same clock -- see step 8

# STEP 6. DELIVERY, AND THE DISCLOSURE.
# SEBI (Intermediaries) (Amendment) Regulations, 2025 -- notified 10 February
# 2025 -- inserted Regulation 16C: a regulated entity is SOLELY liable for
# the AI/ML tools it uses, whether built in-house or bought. Liability covers
# data privacy and security, the integrity of the output, and compliance with
# all applicable law. There is no vendor to point at.

def deliver(advice, client, ai_used):
    return {
        "advice_id": advice["id"],
        "delivered_at_ist": now_ist(),
        "ai_disclosure": describe_ai_use(ai_used) if ai_used else None,
        # Disclosing the EXTENT of AI use is the requirement. "We use AI" is
        # not an extent. Which step, on what inputs, reviewed by whom.
        "reviewed_by": advice["qualified_reviewer"],   # a named person
        "rationale_plain_words": advice["rationale"],
        "rendered_document": advice["pdf_bytes"],      # what they SAW
    }

# STEP 7. THE WALL.
# An investment adviser may not execute. An Execution Only Platform may not
# advise. The client crosses between them; your code must not.

EOP_CATEGORY = {
    1: {"registers_with": "AMFI", "agent_of": "AMC",      "paid_by": "AMC"},
    2: {"registers_with": "stock_exchange_EOP_segment",
        "agent_of": "investor", "paid_by": "investor",
        "base_minimum_capital_rupees": 10_00_000},
}

def handoff(advice, client, eop):
    # Direct plans only. An EOP may not touch regular plans, and an entity may
    # not be Category 1 and Category 2 at once.
    assert all(f["plan"] == "DIRECT" for f in advice["included"]), "EOP: direct plans only"
    assert eop["entity_id"] != advice["adviser_entity_id"], "advice and execution: separate entities"
    return {
        "action": "present_to_client_for_authorisation",   # never auto-execute
        "client_authorises": True,
        "route": eop["entity_id"],
        "adviser_receives_from_amc": 0,        # fee-only. No commission, ever.
    }

# STEP 8. TWO CLOCKS, AND THEY ARE NOT THE SAME CLOCK.
def clocks(client, advice):
    return {
        "profile_valid_until": client["profiled_on"] + timedelta(days=365),
        "advice_valid_until": advice["as_of"] + timedelta(days=advice["shelf_life_days"]),
        # Suitability can lapse with NOBODY transacting: the fund changed
        # mandate, or the client's circumstances did. Both need a trigger.
        "watch": ["mandate_change", "life_event", "drawdown_breach"],
    }

def monitoring_record(run):
    # Record the runs that found NOTHING. An absence of records is
    # indistinguishable from an absence of monitoring, including to you.
    return {"ran_at_ist": now_ist(), "clients_checked": run["n"],
            "exceptions": run["exceptions"], "no_exception_is_a_result": True}

# WHAT TO CHECK
# [ ] the adviser entity receives nothing from any product manufacturer. Not
#     a rebate, not a marketing fee, not a platform contribution
# [ ] no auto-execution. The client authorises, every time
# [ ] a fund changing mandate re-checks EVERY client holding it. Suitability
#     regresses without anyone transacting
# [ ] glide paths beat annual reviews -- a yearly step-down is wrong for 364
#     days and most wrong in the year before the client needs the money
# [ ] the AI disclosure names the STEP and the INPUTS, not the vendor
# [ ] the annual compliance audit is LINE-WISE against every provision, by a
#     CA, CS or CMA. Build the evidence as you go or reconstruct it in March