# app.py
# MBB-style chord diagram (mixed node order) for Mutual Fund churn
# Uses D3 chord layout in browser, static layout (no physics). Mobile-friendly.
import gradio as gr
import pandas as pd
import networkx as nx
import numpy as np
import json
from collections import defaultdict
# -------------------------
# DATA (same as before)
# -------------------------
AMCS = [
"SBI MF", "ICICI Pru MF", "HDFC MF", "Nippon India MF", "Kotak MF",
"UTI MF", "Axis MF", "Aditya Birla SL MF", "Mirae MF", "DSP MF"
]
COMPANIES = [
"HDFC Bank", "ICICI Bank", "Bajaj Finance", "Bajaj Finserv", "Adani Ports",
"Tata Motors", "Shriram Finance", "HAL", "TCS", "AU Small Finance Bank",
"Pearl Global", "Hindalco", "Tata Elxsi", "Cummins India", "Vedanta"
]
BUY_MAP = {
"SBI MF": ["Bajaj Finance", "AU Small Finance Bank"],
"ICICI Pru MF": ["HDFC Bank"],
"HDFC MF": ["Tata Elxsi", "TCS"],
"Nippon India MF": ["Hindalco"],
"Kotak MF": ["Bajaj Finance"],
"UTI MF": ["Adani Ports", "Shriram Finance"],
"Axis MF": ["Tata Motors", "Shriram Finance"],
"Aditya Birla SL MF": ["AU Small Finance Bank"],
"Mirae MF": ["Bajaj Finance", "HAL"],
"DSP MF": ["Tata Motors", "Bajaj Finserv"]
}
SELL_MAP = {
"SBI MF": ["Tata Motors"],
"ICICI Pru MF": ["Bajaj Finance", "Adani Ports"],
"HDFC MF": ["HDFC Bank"],
"Nippon India MF": ["Hindalco"],
"Kotak MF": ["AU Small Finance Bank"],
"UTI MF": ["Hindalco", "TCS"],
"Axis MF": ["TCS"],
"Aditya Birla SL MF": ["Adani Ports"],
"Mirae MF": ["TCS"],
"DSP MF": ["HAL", "Shriram Finance"]
}
COMPLETE_EXIT = {"DSP MF": ["Shriram Finance"]}
FRESH_BUY = {"HDFC MF": ["Tata Elxsi"], "UTI MF": ["Adani Ports"], "Mirae MF": ["HAL"]}
def sanitize_map(m):
out = {}
for k, vals in m.items():
out[k] = [v for v in vals if v in COMPANIES]
return out
BUY_MAP = sanitize_map(BUY_MAP)
SELL_MAP = sanitize_map(SELL_MAP)
COMPLETE_EXIT = sanitize_map(COMPLETE_EXIT)
FRESH_BUY = sanitize_map(FRESH_BUY)
# -------------------------
# Inferred AMC->AMC transfers (same heuristic)
# -------------------------
def infer_amc_transfers(buy_map, sell_map):
transfers = defaultdict(int)
c2s = defaultdict(list)
c2b = defaultdict(list)
for amc, comps in sell_map.items():
for c in comps:
c2s[c].append(amc)
for amc, comps in buy_map.items():
for c in comps:
c2b[c].append(amc)
for c in set(c2s.keys()) | set(c2b.keys()):
for s in c2s[c]:
for b in c2b[c]:
transfers[(s,b)] += 1
return transfers
transfer_counts = infer_amc_transfers(BUY_MAP, SELL_MAP)
# -------------------------
# Build mixed ordering (AMC, company, AMC, company...)
# -------------------------
def build_mixed_ordering(amcs, companies):
mixed = []
n = max(len(amcs), len(companies))
for i in range(n):
if i < len(amcs):
mixed.append(amcs[i])
if i < len(companies):
mixed.append(companies[i])
return mixed
NODES = build_mixed_ordering(AMCS, COMPANIES)
# Node types map for styling
NODE_TYPE = {n: ("amc" if n in AMCS else "company") for n in NODES}
# -------------------------
# Build flow matrix: nodes x nodes
# Matrix interpretation:
# - AMC -> Company for BUY
# - Company -> AMC for SELL
# - AMC -> AMC for inferred TRANSFER
# Fresh buy and complete exit use higher weight
# -------------------------
def build_flow_matrix(nodes):
idx = {n:i for i,n in enumerate(nodes)}
n = len(nodes)
M = [[0]*n for _ in range(n)]
# buys: AMC -> Company
for amc, comps in BUY_MAP.items():
for c in comps:
if amc in idx and c in idx:
w = 1
if amc in FRESH_BUY and c in FRESH_BUY.get(amc, []):
w = 3
M[idx[amc]][idx[c]] += w
# sells: Company -> AMC
for amc, comps in SELL_MAP.items():
for c in comps:
if amc in idx and c in idx:
w = 1
if amc in COMPLETE_EXIT and c in COMPLETE_EXIT.get(amc, []):
w = 3
# represent sell as company -> amc
M[idx[c]][idx[amc]] += w
# inferred transfers: AMC -> AMC
for (s,b), w in transfer_counts.items():
if s in idx and b in idx:
M[idx[s]][idx[b]] += w
return M
MATRIX = build_flow_matrix(NODES)
# -------------------------
# Helper summaries (unchanged)
# -------------------------
def company_trade_summary(company):
buyers = [a for a, cs in BUY_MAP.items() if company in cs]
sellers = [a for a, cs in SELL_MAP.items() if company in cs]
fresh = [a for a, cs in FRESH_BUY.items() if company in cs]
exits = [a for a, cs in COMPLETE_EXIT.items() if company in cs]
df = pd.DataFrame({
"Role": (["Buyer"] * len(buyers)) + (["Seller"] * len(sellers)) + (["Fresh buy"] * len(fresh)) + (["Complete exit"] * len(exits)),
"AMC": buyers + sellers + fresh + exits
})
if df.empty:
return None, pd.DataFrame([], columns=["Role","AMC"])
counts = df.groupby("Role").size().reset_index(name="Count")
fig = {
"data": [{"type":"bar","x": counts["Role"].tolist(), "y": counts["Count"].tolist()}],
"layout":{"title":f"Trades for {company}"}
}
return fig, df
def amc_transfer_summary(amc):
sold = SELL_MAP.get(amc, [])
transfers = []
for s in sold:
buyers = [a for a, cs in BUY_MAP.items() if s in cs]
for b in buyers:
transfers.append({"security": s, "buyer_amc": b})
df = pd.DataFrame(transfers)
if df.empty:
return None, pd.DataFrame([], columns=["security","buyer_amc"])
counts = df["buyer_amc"].value_counts().reset_index()
counts.columns = ["Buyer AMC","Count"]
fig = {
"data": [{"type":"bar","x": counts["Buyer AMC"].tolist(), "y": counts["Count"].tolist()}],
"layout":{"title":f"Inferred transfers from {amc}"}
}
return fig, df
# -------------------------
# Build HTML with D3 chord
# -------------------------
def make_chord_html(nodes, matrix, node_type):
nodes_json = json.dumps(nodes)
mat_json = json.dumps(matrix)
types_json = json.dumps(node_type)
# D3 chord diagram: mixed nodes around circle, modern palette
html = f"""
Legend
AMC nodes
Company nodes Note: TRANSFER connections are inferred from simultaneous buys/sells, not explicitly reported.