"""Auditable evidence profiles and population-data QC. Python 3.12, stdlib only. Import assess(), audit_csv() and export_audit_csv() from this module. CSV values are observations, not inputs to a clinical risk score. """ from collections import Counter, defaultdict from datetime import datetime, timezone from decimal import Decimal, InvalidOperation from pathlib import Path import argparse import csv import hashlib import io import json VERSION = '1.2.0' ROUNDING_TOLERANCE = Decimal('0.0005') REQUIRED = {'rsid', 'target_gene', 'population_code', 'dataset', 'source_url', 'assembly', 'focus_allele', 'allele_count', 'allele_number_derived', 'allele_frequency_source', 'is_russian_cohort'} AUDIT_FIELDS = ('audit_status', 'audit_flags', 'allele_frequency_recomputed', 'audit_frequency_difference', 'frequency_tolerance', 'source_row') SOURCE_NOTICE = ('Библиографические ссылки приведены для проверки утверждений. ' 'Полные тексты не проходили новую независимую экспертизу при подготовке этого раздела.') SOURCES = { 'galnac': {'title': 'Nair et al., 2014. Multivalent GalNAc–siRNA conjugates', 'url': 'https://doi.org/10.1021/ja505986a', 'scope': 'Доставка GalNAc–siRNA в гепатоциты; не доказательство действия конкретной siRNA к F11.'}, 'rna': {'title': 'Dowdy, 2017. Overcoming cellular barriers for RNA therapeutics', 'url': 'https://doi.org/10.1038/nbt.3802', 'scope': 'Обзор барьеров клеточной доставки РНК.'}, 'fxi_aso': {'title': 'Büller et al., 2015. Factor XI antisense oligonucleotide', 'url': 'https://doi.org/10.1056/NEJMoa1405760', 'scope': 'Исследование ASO к F11. ASO и siRNA — разные модальности; результаты нельзя приписывать siRNA.'}, 'fxi_ab': {'title': 'Verhamme et al., 2021. Abelacimab for prevention of venous thromboembolism', 'url': 'https://doi.org/10.1056/NEJMoa2105872', 'scope': 'Конкретное антитело и изученный клинический контекст, не универсальная безопасность ингибирования F11.'}, 'glue': {'title': 'Matyskiela et al., 2016. Recruitment of GSPT1 to CRL4–CRBN', 'url': 'https://doi.org/10.1038/nature18611', 'scope': 'Механизм CC-885. Не обосновывает эффективность нового деградатора или любого препарата к GSPT1.'}, 'editing': {'title': 'FDA, 2024. Human Gene Therapy Products Incorporating Human Genome Editing', 'url': 'https://www.fda.gov/regulatory-information/search-fda-guidance-documents/human-gene-therapy-products-incorporating-human-genome-editing', 'scope': 'Требования к оценке продуктов редактирования. Не одобрение редактирования выбранного гена.'}, 'f7': {'title': 'EMA. NovoSeven: European public assessment report', 'url': 'https://www.ema.europa.eu/en/medicines/human/EPAR/novoseven', 'scope': 'Рекомбинантный активированный фактор VII в конкретных показаниях. Не свидетельство безопасности снижения F7.'}, 'myc': {'title': 'ClinicalTrials.gov. NCT04808362 — OMO-103', 'url': 'https://clinicaltrials.gov/study/NCT04808362', 'scope': 'Реестровая запись конкретного исследования; регистрация не доказывает пользу или эффективность.'}, } STATUS = {'example': 'Опубликованный пример', 'platform': 'Данные платформы', 'context': 'Косвенный контекст', 'unresolved': 'Требует проверки', 'missing': 'Нет данных в этом наборе'} def evidence(domain, status, claim, boundary, *sources): return dict(domain=domain, status=status, status_label=STATUS[status], claim=claim, boundary=boundary, sources=list(sources)) COMMON = [ evidence('Популяционные данные', 'unresolved', 'Частоты требуют проверки качества и соответствия конкретному аллелю.', 'Частота варианта не равна распространённости заболевания, эффекту препарата или безопасности.'), evidence('Этика и безопасность', 'missing', 'В CSV нет данных о безопасности, согласии участников или допустимом соотношении пользы и риска.', 'Оценка зависит от заболевания, популяции, обратимости вмешательства и протокола; универсального этического балла нет.'), ] PROFILES = { 'F11:sirna': dict(gene='F11', modality='sirna', label='F11 · siRNA', question='Можно ли обосновать последовательностно-зависимое снижение F11 в гепатоцитах?', evidence=[ evidence('Механизм', 'context', 'Для F11 существует опубликованный пример снижения синтеза с помощью ASO.', 'Это поддерживает рассмотрение мишени, но не подтверждает эффективность siRNA.', 'fxi_aso'), evidence('Доставка', 'platform', 'GalNAc-конъюгация используется для доставки siRNA в гепатоциты.', 'Нужны данные конкретной конструкции, продуктивного выхода из эндосом и внутриклеточной активности.', 'galnac', 'rna'), evidence('Клиническая применимость', 'unresolved', 'Приведённые источники не устанавливают клиническую пользу выбранной siRNA.', 'Нельзя переносить результаты ASO или антитела на новую siRNA.', 'fxi_aso', 'fxi_ab')], next_steps=['Уточнить заболевание, требуемое воздействие и критерий результата.', 'Проверить транскрипт, последовательность, специфичность и химическую конструкцию.', 'Измерить мРНК F11, белок и активность FXI с контрольной последовательностью.', 'Отдельно оценить доставку, нежелательные эффекты и функциональный результат.']), 'F11:antibody': dict(gene='F11', modality='antibody', label='F11 · антитело', question='Поддерживают ли данные внеклеточное воздействие на фактор XI выбранным антителом?', evidence=[ evidence('Механизм', 'example', 'Abelacimab — опубликованный пример антитела к FXI/FXIa.', 'Вывод ограничен конкретной молекулой и изученным контекстом.', 'fxi_ab'), evidence('Доставка', 'context', 'Для взаимодействия с циркулирующим FXI не требуется проникновение в гепатоцит.', 'Экспозиция, эпитоп, связывание и функция проверяются для конкретного антитела.', 'fxi_ab'), evidence('Клиническая применимость', 'example', 'Для abelacimab опубликовано клиническое исследование.', 'Это не рекомендация к разработке другого антитела и не утверждение универсальной безопасности.', 'fxi_ab')], next_steps=['Определить эпитоп, состояние белка и предполагаемый механизм ингибирования.', 'Разделить подтверждение связывания и функционального действия.', 'Обосновать экспозицию, селективность и применимость клинического контекста.']), 'F11:editing': dict(gene='F11', modality='editing', label='F11 · редактирование', question='Есть ли отдельное обоснование устойчивого изменения F11, помимо обратимого подавления?', evidence=[ evidence('Механизм', 'context', 'Фармакологические подходы к F11 дают контекст для рассмотрения мишени.', 'Эффекты лекарственного подавления не доказывают безопасность постоянного редактирования.', 'fxi_aso', 'fxi_ab'), evidence('Доставка и контроль', 'unresolved', 'Нужны данные выбранного редактора, системы доставки и распределения.', 'Наличие GalNAc–siRNA не доказывает возможность доставки редактора.', 'editing'), evidence('Долгосрочная безопасность', 'unresolved', 'Нужны оценка непреднамеренных изменений и план длительного наблюдения.', 'В загруженном наборе таких данных нет.', 'editing')], next_steps=['Обосновать необходимость необратимого вмешательства по сравнению с альтернативами.', 'Определить редактор, целевые клетки, механизм и длительность воздействия.', 'Составить план проверки внецелевых и непреднамеренных изменений.', 'Провести отдельную регуляторную и этическую оценку.']), 'F7:lowering': dict(gene='F7', modality='lowering', label='F7 · снижение активности', question='Обосновано ли снижение F7 для выбранного заболевания?', evidence=[ evidence('Направление воздействия', 'unresolved', 'Клиническое применение активированного фактора VII не доказывает пользу его снижения.', 'Замещение и подавление — противоположные стратегии; смешивать их нельзя.', 'f7'), evidence('Гемостаз', 'unresolved', 'Нужно определить допустимую степень и длительность снижения функции.', 'Популяционная частота варианта не задаёт безопасную дозу или терапевтическое окно.', 'f7')], next_steps=['Уточнить заболевание и обосновать направление воздействия.', 'Проверить публикации и доступность счётчиков для конкретных вариантов F7; метка когорты не означает репрезентативность РФ.', 'Спланировать функциональные тесты гемостаза и оценки безопасности.']), 'MYC:inhibition': dict(gene='MYC', modality='inhibition', label='MYC · ингибирование', question='Какой механизм ингибирования MYC обоснован в конкретной опухолевой модели?', evidence=[ evidence('Пример разработки', 'context', 'OMO-103 представлен в реестре клинических исследований.', 'Реестровая запись не является доказательством эффективности или результатом исследования.', 'myc'), evidence('Механизм и применимость', 'unresolved', 'Нужно уточнить тип опухоли, экспрессию, зависимость и способ воздействия.', 'Герминальные популяционные частоты не заменяют соматические изменения или функциональную зависимость опухоли.', 'myc')], next_steps=['Определить опухолевый контекст и биомаркер зависимости от MYC.', 'Проверить опубликованные результаты конкретной модальности.', 'Разделить подтверждение взаимодействия, клеточного эффекта и селективности.']), 'GSPT1:glue': dict(gene='GSPT1', modality='glue', label='GSPT1 · молекулярный клей', question='Подтверждается ли CRBN-зависимая деградация GSPT1 выбранным соединением?', evidence=[ evidence('Механизм', 'example', 'CC-885 описан как пример рекрутирования GSPT1 комплексом CRL4–CRBN.', 'Не каждая малая молекула или лиганд CRBN рекрутирует GSPT1.', 'glue'), evidence('Клеточный эффект', 'unresolved', 'Нужны внутриклеточная экспозиция, зависимость от CRBN и протеасомы.', 'Снижение количества белка следует отличать от подавления синтеза или общего повреждения клеток.', 'glue'), evidence('Селективность', 'unresolved', 'Нужны оценка других субстратов и проверка причинной роли GSPT1.', 'Опубликованный механизм не устанавливает терапевтическое окно нового соединения.', 'glue')], next_steps=['Проверить экспозицию и взаимодействие с релевантными белками.', 'Измерить GSPT1 во времени с контролями CRBN и протеасомы.', 'Проверить другие субстраты и причинную связь с фенотипом.', 'Отдельно оценить селективность между выбранной моделью и нормальными клетками.']), } def _number(value): try: result = Decimal(str(value).strip()) return result if result.is_finite() else None except (InvalidOperation, ValueError): return None def audit_csv(path, tolerance=ROUNDING_TOLERANCE): """Read source values unchanged; append computed fields and explicit QC flags. No counts are reconstructed from frequencies. Zero/zero means unavailable. Tolerance is an absolute difference, not a statistical confidence interval. """ tolerance = _number(tolerance) if tolerance is None or not Decimal(0) <= tolerance <= Decimal(1): raise ValueError('Допуск должен быть конечным числом от 0 до 1.') path = Path(path) content = path.read_bytes() reader = csv.DictReader(io.StringIO(content.decode('utf-8-sig'), newline='')) fields = reader.fieldnames or [] if len(set(fields)) != len(fields) or not REQUIRED.issubset(fields): raise ValueError('CSV: отсутствуют обязательные поля или повторяются имена столбцов.') if set(fields).intersection(AUDIT_FIELDS): raise ValueError('Нужен исходный CSV без добавленных audit-полей.') rows = list(reader) if not rows: raise ValueError('CSV не содержит строк данных.') groups = defaultdict(list) for line, row in enumerate(rows, 2): if None in row or any(v is None for v in row.values()): raise ValueError(f'Неверное число полей в строке {line}.') flags = [] frequency = _number(row['allele_frequency_source']) ac, an = _number(row['allele_count']), _number(row['allele_number_derived']) recomputed = difference = None if frequency is None or not 0 <= frequency <= 1: flags.append('invalid_source_frequency') if (ac == 0 and an == 0) or row['allele_count'].strip()=='' or row['allele_number_derived'].strip()=='': flags.append('missing_counts') elif ac is None or an is None or ac != ac.to_integral_value() or an != an.to_integral_value() or an <= 0 or not 0 <= ac <= an: flags.append('invalid_counts') else: recomputed = ac / an if frequency is not None and 0 <= frequency <= 1: difference = recomputed - frequency if abs(difference) > tolerance: flags.append('frequency_mismatch') if 'allele_frequency_recalculated' in row: declared = _number(row['allele_frequency_recalculated']) if row['allele_frequency_recalculated'].strip() and declared is None: flags.append('invalid_declared_recalculation') if declared is not None and recomputed is None: flags.append('declared_recalculation_unverifiable') elif declared is not None and recomputed is not None and abs(declared - recomputed) > tolerance: flags.append('declared_recalculation_mismatch') if row.get('allele_number_status') == 'derived_not_source_reported': flags.append('derived_denominator') if row['is_russian_cohort'].strip().lower() not in {'true', 'false'}: flags.append('invalid_cohort_flag') row.update(audit_status='requires_review' if flags else 'arithmetically_consistent', audit_flags=flags, allele_frequency_recomputed=str(recomputed) if recomputed is not None else None, audit_frequency_difference=str(difference) if difference is not None else None, frequency_tolerance=str(tolerance), source_row=line) key = tuple(row.get(k, '') for k in ('rsid', 'target_gene', 'assembly', 'population_code', 'focus_allele', 'dataset', 'source_url', 'retrieved_utc')) groups[key].append(row) ambiguous = 0 for group in groups.values(): if len({_number(row['allele_frequency_source']) for row in group}) > 1: ambiguous += 1 for row in group: row['audit_flags'].append('ambiguous_allele_frequency') row['audit_status'] = 'requires_review' flags = Counter(flag for row in rows for flag in row['audit_flags']) def summarize(selected): russian = [row for row in selected if row['is_russian_cohort'].strip().lower() == 'true'] return dict(rows=len(selected), variants=len({row['rsid'] for row in selected}), population_codes=len({row['population_code'] for row in selected}), russian_rows=len(russian), russian_variants=len({row['rsid'] for row in russian}), russian_cohort_labels=len({(row['population_code'], row['dataset'], row['source_url']) for row in russian}), missing_counts=sum('missing_counts' in row['audit_flags'] for row in selected), mismatched_frequencies=sum('frequency_mismatch' in row['audit_flags'] for row in selected), derived_denominators=sum('derived_denominator' in row['audit_flags'] for row in selected), invalid_counts=sum('invalid_counts' in row['audit_flags'] for row in selected), invalid_frequencies=sum('invalid_source_frequency' in row['audit_flags'] for row in selected)) return dict(version=VERSION, source_sha256=hashlib.sha256(content).hexdigest(), original_fields=fields, tolerance=str(tolerance), summary=summarize(rows), by_gene={gene: summarize([row for row in rows if row['target_gene'] == gene]) for gene in sorted({row['target_gene'] for row in rows})}, source_labels=dict(Counter(row['dataset'] for row in rows)), primary_source_labels=dict(Counter(row.get('primary_dataset') or row.get('source_id') or row['dataset'] for row in rows)), denominator_statuses=dict(Counter(row.get('allele_number_status') or 'not_declared' for row in rows)), flags=dict(flags), ambiguous_allele_groups=ambiguous, rows=rows, limitations=['Согласованность AC/AN не подтверждает достоверность исходной публикации. AN, восстановленный из возвращённых AC, не является сообщённым источником знаменателем; полнота набора аллелей отдельно не подтверждена.', 'Метки популяций и когорт не означают независимые выборки; их пересечения не исключены.', 'Исходные частоты и счётчики сохранены. При расхождении нельзя автоматически выбрать правильное значение.']) def index_variants(audit): """Navigation index from raw allele rows; min(f,1-f) is not a validated MAF.""" grouped=defaultdict(list) for row in audit['rows']: grouped[(row['target_gene'],row['rsid'])].append(row) result=[] for (gene,rsid),rows in sorted(grouped.items(),key=lambda x:(x[0][0],int(x[0][1][2:]) if x[0][1][2:].isdigit() else 0)): values=[min(float(r['allele_frequency_source']),1-float(r['allele_frequency_source'])) for r in rows if 'invalid_source_frequency' not in r['audit_flags']] first=rows[0] result.append(dict(rsid=rsid,gene=gene,assembly=first['assembly'], release=first.get('ensembl_release') or 'не указан',retrieved=max(r.get('retrieved_utc','') for r in rows), sources=sorted({r['dataset'] for r in rows}),url=first['source_url'], npop=len({r['population_code'] for r in rows}),mmax=max(values,default=0.0), poly=max(values,default=0.0)>=0.01,ru=any(r['is_russian_cohort'].lower()=='true' for r in rows), chromosome=first.get('chromosome',''),position=first.get('position',''), reference_allele=first.get('reference_allele',''), consequence=first.get('most_severe_consequence',''), clinical_labels_site_level=first.get('clinical_labels_site_level',''))) return result def export_audit_csv(audit): stream = io.StringIO(newline='') writer = csv.DictWriter(stream, audit['original_fields'] + list(AUDIT_FIELDS), extrasaction='ignore') writer.writeheader() for row in audit['rows']: writer.writerow(dict(row, audit_flags=';'.join(row['audit_flags']))) return '\ufeff' + stream.getvalue() def assess(gene, modality, audit=None): """Build an evidence dossier; deliberately returns no risk score or recommendation.""" key = f'{str(gene).strip().upper()}:{str(modality).strip().lower()}' if key not in PROFILES: raise ValueError('Для этой пары мишень–модальность нет подготовленного профиля.') profile = json.loads(json.dumps(PROFILES[key], ensure_ascii=False)) profile['evidence'] += json.loads(json.dumps(COMMON, ensure_ascii=False)) used = {ref for item in profile['evidence'] for ref in item['sources']} profile.update(version=VERSION, generated_utc=datetime.now(timezone.utc).isoformat(), conclusion='Профиль обоснования гипотезы; решение о разработке не вычисляется.', sources={key: dict(SOURCES[key]) for key in sorted(used)}, source_notice=SOURCE_NOTICE, population=None, data_quality=None) if audit is not None: profile['population'] = audit['by_gene'].get(profile['gene']) profile['data_quality'] = {key: audit[key] for key in ['source_sha256', 'summary', 'flags', 'ambiguous_allele_groups', 'tolerance', 'limitations', 'source_labels', 'primary_source_labels', 'denominator_statuses']} profile['data_quality']['mismatch_examples'] = [{key: row[key] for key in ['rsid', 'target_gene', 'population_code', 'focus_allele', 'allele_count', 'allele_number_derived', 'allele_frequency_source', 'allele_frequency_recomputed', 'source_row']} for row in audit['rows'] if 'frequency_mismatch' in row['audit_flags']][:20] profile['traffic_lights'] = traffic_lights(profile) return profile def traffic_lights(report): """Three separate evidence signals, never a combined clinical go/no-go.""" def signal(domain, item): color = 'green' if item['status'] == 'example' else 'red' if item['status'] == 'missing' else 'yellow' return dict(domain=domain, color=color, label=item['status_label'], reason=item['claim'], limitation=item['boundary'], sources=list(item['sources'])) evidence = report['evidence'] lights = [signal('Научное обоснование', evidence[0]), signal('Реализуемость подхода', evidence[1])] quality = report.get('population') if not quality or not quality['rows']: color, label, reason = 'red', 'Нет проверяемого набора', 'Данные выбранной мишени не подключены.' elif quality['missing_counts'] or quality['mismatched_frequencies'] or quality.get('invalid_counts') or quality.get('invalid_frequencies'): color, label = 'red', 'Частоты требуют проверки' reason = 'Интерпретация данных: полнота и согласованность исходных наблюдений требуют проверки.' else: color, label, reason = 'yellow', 'Нужна проверка первоисточника', 'Доступность счётчиков не подтверждает достоверность и применимость выборки.' lights.append(dict(domain='Безопасность', color=color, label=label, reason=reason, limitation='Это состояние данных, не риск лечения и не патогенность варианта.', sources=[])) return dict(signals=lights, legend={ 'green': 'Есть опубликованный пример для указанного механизма; перенос на другой препарат не подтверждён.', 'yellow': 'Есть косвенный контекст или необходима дополнительная проверка.', 'red': 'Нет проверяемых данных либо есть препятствие для их интерпретации; это не противопоказание.'}, conclusion='Общий цвет, рейтинг безопасности и рекомендация к разработке не вычисляются.') def catalog(): return dict(version=VERSION, profiles=[dict(id=key, gene=value['gene'], modality=value['modality'], label=value['label']) for key, value in PROFILES.items()], status_labels=STATUS) def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('csv', type=Path) parser.add_argument('--gene', default='F11') parser.add_argument('--modality', default='sirna') parser.add_argument('--audit-output', type=Path) parser.add_argument('--report-output', type=Path) args = parser.parse_args() outputs = [p.resolve() for p in [args.audit_output, args.report_output] if p is not None] if args.csv.resolve() in outputs or len(outputs) != len(set(outputs)): parser.error('Выходные файлы должны отличаться друг от друга и от исходного CSV.') audit = audit_csv(args.csv) report = assess(args.gene, args.modality, audit) if args.audit_output: if args.audit_output.resolve() == args.csv.resolve(): parser.error('Нельзя перезаписывать исходный CSV.') args.audit_output.write_text(export_audit_csv(audit), encoding='utf-8') result = json.dumps(report, ensure_ascii=False, indent=2) if args.report_output: if args.report_output.resolve() == args.csv.resolve(): parser.error('Нельзя перезаписывать исходный CSV.') args.report_output.write_text(result, encoding='utf-8') else: print(result) if __name__ == '__main__': main()