Week 8: Domain Projects

@prefix med: <http://example.org/medical#> .
@prefix snomed: <http://snomed.info/id/> .
@prefix fhir: <http://hl7.org/fhir/> .
 
# Disease class hierarchy
med:Disease a owl:Class .
med:CardiovascularDisease rdfs:subClassOf med:Disease .
med:HeartFailure rdfs:subClassOf med:CardiovascularDisease ;
    owl:equivalentClass snomed:84114007 .
 
# Treatment relationships
med:treats a owl:ObjectProperty ;
    rdfs:domain med:Treatment ;
    rdfs:range med:Disease .
 
med:contraindicatedWith a owl:ObjectProperty ;
    rdfs:domain med:Drug ;
    rdfs:range med:Condition .

from rdflib import Graph, Namespace, Literal
from rdflib.namespace import RDF, RDFS, OWL
 
class ClinicalKnowledgeGraph:
    def __init__(self):
        self.g = Graph()
        self.MED = Namespace("http://example.org/medical#")
        self.SNOMED = Namespace("http://snomed.info/id/")
 
        self.g.bind("med", self.MED)
        self.g.bind("snomed", self.SNOMED)
 
    def add_condition(self, code, name, parent=None):
        """Add a medical condition with SNOMED mapping."""
        condition_uri = self.MED[f"condition_{code}"]
 
        self.g.add((condition_uri, RDF.type, self.MED.Condition))
        self.g.add((condition_uri, RDFS.label, Literal(name)))
        self.g.add((condition_uri, OWL.sameAs, self.SNOMED[code]))
 
        if parent:
            parent_uri = self.MED[f"condition_{parent}"]
            self.g.add((condition_uri, RDFS.subClassOf, parent_uri))
 
    def add_treatment(self, drug_name, treats_condition, contraindications=None):
        """Add treatment information."""
        drug_uri = self.MED[drug_name.replace(" ", "_")]
 
        self.g.add((drug_uri, RDF.type, self.MED.Drug))
        self.g.add((drug_uri, RDFS.label, Literal(drug_name)))
        self.g.add((drug_uri, self.MED.treats,
                    self.MED[f"condition_{treats_condition}"]))
 
        if contraindications:
            for contra in contraindications:
                self.g.add((drug_uri, self.MED.contraindicatedWith,
                           self.MED[f"condition_{contra}"]))
 
    def check_drug_interactions(self, drug_list):
        """Query for potential drug interactions."""
        query = """
        PREFIX med: <http://example.org/medical#>
 
        SELECT ?drug1 ?drug2 ?condition
        WHERE {
            ?drug1 med:contraindicatedWith ?condition .
            ?drug2 med:treats ?condition .
            FILTER(?drug1 != ?drug2)
        }
        """
        results = self.g.query(query)
        return [(row.drug1, row.drug2, row.condition) for row in results]

class ClinicalDecisionSupport:
    def __init__(self, knowledge_graph):
        self.kg = knowledge_graph
 
    def recommend_treatment(self, diagnosis, patient_conditions):
        """Recommend treatments considering contraindications."""
        query = f"""
        PREFIX med: <http://example.org/medical#>
 
        SELECT ?treatment ?efficacy
        WHERE {{
            ?treatment med:treats med:condition_{diagnosis} .
            ?treatment med:efficacy ?efficacy .
 
            # Exclude contraindicated treatments
            FILTER NOT EXISTS {{
                ?treatment med:contraindicatedWith ?contra .
                VALUES ?contra {{ {' '.join([f'med:condition_{c}' for c in patient_conditions])} }}
            }}
        }}
        ORDER BY DESC(?efficacy)
        """
 
        results = self.kg.g.query(query)
        return list(results)

@prefix legal: <http://example.org/legal#> .
@prefix lkif: <http://www.estrellaproject.org/lkif-core/> .
 
# Legal document hierarchy
legal:LegalDocument a owl:Class .
legal:Statute rdfs:subClassOf legal:LegalDocument .
legal:CaseLaw rdfs:subClassOf legal:LegalDocument .
legal:Regulation rdfs:subClassOf legal:LegalDocument .
 
# Relationships
legal:cites a owl:ObjectProperty ;
    rdfs:domain legal:LegalDocument ;
    rdfs:range legal:LegalDocument .
 
legal:overrules a owl:ObjectProperty ;
    rdfs:subPropertyOf legal:cites ;
    rdfs:domain legal:CaseLaw ;
    rdfs:range legal:CaseLaw .
 
legal:jurisdiction a owl:ObjectProperty ;
    rdfs:domain legal:LegalDocument ;
    rdfs:range legal:Jurisdiction .

class LegalKnowledgeGraph:
    def __init__(self):
        self.g = Graph()
        self.LEGAL = Namespace("http://example.org/legal#")
        self.g.bind("legal", self.LEGAL)
 
    def add_case(self, case_id, title, date, jurisdiction, topic):
        """Add a court case to the knowledge graph."""
        case_uri = self.LEGAL[case_id]
 
        self.g.add((case_uri, RDF.type, self.LEGAL.CaseLaw))
        self.g.add((case_uri, self.LEGAL.title, Literal(title)))
        self.g.add((case_uri, self.LEGAL.date, Literal(date)))
        self.g.add((case_uri, self.LEGAL.jurisdiction,
                    self.LEGAL[jurisdiction]))
        self.g.add((case_uri, self.LEGAL.topic, self.LEGAL[topic]))
 
    def add_citation(self, citing_case, cited_case, citation_type="cites"):
        """Add citation relationship between cases."""
        citing_uri = self.LEGAL[citing_case]
        cited_uri = self.LEGAL[cited_case]
 
        predicate = self.LEGAL[citation_type]
        self.g.add((citing_uri, predicate, cited_uri))
 
    def find_precedents(self, topic, jurisdiction):
        """Find relevant precedent cases."""
        query = f"""
        PREFIX legal: <http://example.org/legal#>
 
        SELECT ?case ?title ?date (COUNT(?citation) AS ?authority)
        WHERE {{
            ?case a legal:CaseLaw ;
                  legal:topic legal:{topic} ;
                  legal:jurisdiction legal:{jurisdiction} ;
                  legal:title ?title ;
                  legal:date ?date .
 
            OPTIONAL {{
                ?other_case legal:cites ?case .
                BIND(?other_case AS ?citation)
            }}
        }}
        GROUP BY ?case ?title ?date
        ORDER BY DESC(?authority)
        """
        return list(self.g.query(query))
 
    def trace_legal_reasoning(self, case_id):
        """Trace the chain of citations for legal reasoning."""
        query = f"""
        PREFIX legal: <http://example.org/legal#>
 
        SELECT ?citing ?cited ?type
        WHERE {{
            legal:{case_id} legal:cites* ?citing .
            ?citing ?type ?cited .
            FILTER(?type IN (legal:cites, legal:overrules, legal:distinguishes))
        }}
        """
        return list(self.g.query(query))

@prefix fibo: <https://spec.edmcouncil.org/fibo/ontology/> .
@prefix fin: <http://example.org/finance#> .
 
# Financial instruments
fin:FinancialInstrument a owl:Class .
fin:Equity rdfs:subClassOf fin:FinancialInstrument .
fin:Bond rdfs:subClassOf fin:FinancialInstrument .
fin:Derivative rdfs:subClassOf fin:FinancialInstrument .
 
# Organization relationships
fin:isIssuedBy a owl:ObjectProperty ;
    rdfs:domain fin:FinancialInstrument ;
    rdfs:range fin:Organization .
 
fin:isRegulatedBy a owl:ObjectProperty ;
    rdfs:domain fin:Organization ;
    rdfs:range fin:RegulatoryBody .
 
# Risk relationships
fin:hasRiskExposure a owl:ObjectProperty ;
    rdfs:domain fin:Portfolio ;
    rdfs:range fin:RiskFactor .

class FinancialKnowledgeGraph:
    def __init__(self):
        self.g = Graph()
        self.FIN = Namespace("http://example.org/finance#")
        self.g.bind("fin", self.FIN)
 
    def add_company(self, ticker, name, sector, market_cap):
        """Add a company to the knowledge graph."""
        company_uri = self.FIN[ticker]
 
        self.g.add((company_uri, RDF.type, self.FIN.Company))
        self.g.add((company_uri, self.FIN.ticker, Literal(ticker)))
        self.g.add((company_uri, self.FIN.name, Literal(name)))
        self.g.add((company_uri, self.FIN.sector, self.FIN[sector]))
        self.g.add((company_uri, self.FIN.marketCap, Literal(market_cap)))
 
    def add_relationship(self, company1, company2, rel_type, details=None):
        """Add business relationship between companies."""
        c1_uri = self.FIN[company1]
        c2_uri = self.FIN[company2]
 
        rel_uri = self.FIN[f"{rel_type}_{company1}_{company2}"]
        self.g.add((rel_uri, RDF.type, self.FIN[rel_type]))
        self.g.add((rel_uri, self.FIN.from_, c1_uri))
        self.g.add((rel_uri, self.FIN.to, c2_uri))
 
        if details:
            for key, value in details.items():
                self.g.add((rel_uri, self.FIN[key], Literal(value)))
 
    def analyze_supply_chain_risk(self, company):
        """Analyze supply chain risk exposure."""
        query = f"""
        PREFIX fin: <http://example.org/finance#>
 
        SELECT ?supplier ?tier ?risk_score
        WHERE {{
            fin:{company} fin:dependsOn+ ?supplier .
            ?supplier fin:riskScore ?risk_score .
 
            # Calculate tier depth
            {{
                SELECT ?supplier (COUNT(?intermediate) AS ?tier)
                WHERE {{
                    fin:{company} fin:dependsOn* ?intermediate .
                    ?intermediate fin:dependsOn ?supplier .
                }}
                GROUP BY ?supplier
            }}
        }}
        ORDER BY ?tier DESC(?risk_score)
        """
        return list(self.g.query(query))
 
    def find_regulatory_exposure(self, portfolio_companies):
        """Find regulatory bodies affecting portfolio."""
        query = f"""
        PREFIX fin: <http://example.org/finance#>
 
        SELECT ?regulator (COUNT(?company) AS ?exposure)
        WHERE {{
            VALUES ?company {{ {' '.join([f'fin:{c}' for c in portfolio_companies])} }}
            ?company fin:isRegulatedBy ?regulator .
        }}
        GROUP BY ?regulator
        ORDER BY DESC(?exposure)
        """
        return list(self.g.query(query))

class CrossDomainKnowledgeGraph:
    def __init__(self):
        self.medical = ClinicalKnowledgeGraph()
        self.legal = LegalKnowledgeGraph()
        self.finance = FinancialKnowledgeGraph()
 
        # Unified namespace for cross-domain links
        self.XDOMAIN = Namespace("http://example.org/crossdomain#")
 
    def link_pharma_company_to_drugs(self, company_ticker, drug_names):
        """Link financial entity to medical products."""
        company_uri = self.finance.FIN[company_ticker]
 
        for drug in drug_names:
            drug_uri = self.medical.MED[drug.replace(" ", "_")]
            self.finance.g.add((company_uri, self.XDOMAIN.manufactures, drug_uri))
 
    def link_regulatory_case(self, company_ticker, case_id):
        """Link company to regulatory legal case."""
        company_uri = self.finance.FIN[company_ticker]
        case_uri = self.legal.LEGAL[case_id]
 
        self.finance.g.add((company_uri, self.XDOMAIN.involvedIn, case_uri))
 
    def analyze_pharma_risk(self, company_ticker):
        """Analyze pharmaceutical company risk across domains."""
        # Drug safety issues
        drug_issues = self.medical.check_drug_interactions([])
 
        # Regulatory cases
        legal_issues = self.legal.find_precedents("DrugSafety", "FDA")
 
        # Financial exposure
        supply_chain = self.finance.analyze_supply_chain_risk(company_ticker)
 
        return {
            "drug_safety": drug_issues,
            "legal_exposure": legal_issues,
            "supply_chain_risk": supply_chain
        }

class DomainRAG:
    def __init__(self, domain_kg, llm):
        self.kg = domain_kg
        self.llm = llm
 
    def answer(self, question, domain="medical"):
        """Answer domain-specific questions using KG context."""
        # Extract entities from question
        entities = self._extract_entities(question, domain)
 
        # Retrieve relevant subgraph
        context = self._get_kg_context(entities, domain)
 
        # Generate answer
        prompt = f"""
        You are an expert in the {domain} domain.
        Using the following knowledge graph context, answer the question.
 
        Context:
        {context}
 
        Question: {question}
 
        Provide a detailed answer with references to the knowledge graph.
        """
 
        return self.llm.invoke(prompt)
 
    def _get_kg_context(self, entities, domain):
        """Retrieve domain-specific context."""
        if domain == "medical":
            return self._medical_context(entities)
        elif domain == "legal":
            return self._legal_context(entities)
        elif domain == "finance":
            return self._finance_context(entities)
 
    def _medical_context(self, entities):
        """Build medical context from conditions and treatments."""
        context_parts = []
        for entity in entities:
            # Get related treatments
            treatments = self.kg.get_treatments(entity)
            # Get contraindications
            contraindications = self.kg.get_contraindications(entity)
            context_parts.append(f"Condition: {entity}")
            context_parts.append(f"Treatments: {treatments}")
            context_parts.append(f"Contraindications: {contraindications}")
        return "\n".join(context_parts)

(:Disease {name: "Diabetes", icd10: "E11"})
(:Doctor {name: "Dr. Smith", specialty: "Endocrinology"})
(:Treatment {name: "Insulin Therapy"})
 
(:Doctor)-[:SPECIALIZES_IN]->(:Specialty)
(:Doctor)-[:TREATS]->(:Disease)
(:Disease)-[:TREATED_BY]->(:Treatment)