feat: 프라이머 설계 알고리즘 최적화 및 좌표계 관련 수정

app/algorithms/PrimerDesigner.py

@@ -72,7 +72,6 @@ def __init__(self, genome_fasta: str, annotation_db: str):
         self.genome = pysam.FastaFile(genome_fasta)
         self.db = sqlite3.connect(annotation_db)
         self.cur = self.db.cursor()
-
     def generate_candidates(
         self,
         template: str,
@@ -83,7 +82,7 @@ def generate_candidates(
         max_poly_x=4,
         gc_clamp=True,
     ) -> List[Dict]:
-        """Stage 2.1: 물성 기반 후보군 생성"""
+        """Stage 2.1: 물성 기반 후보군 생성 (1-based 반영)"""
         candidates = []
         for k in range(k_min, k_max + 1):
             for i in range(len(template) - k + 1):
@@ -92,14 +91,16 @@ def generate_candidates(
                     tm = calc_tm_nn(s)
                     if not (tm_range[0] <= tm <= tm_range[1]):
                         continue
-                    # 1. Poly-X 필터 반영
+
+                    # 1. Poly-X 필터
                     if any(base * max_poly_x in s for base in "ATCG"):
                         continue
 
-                    # 2. GC Content 필터 반영
+                    # 2. GC Content 필터
                     gc_content = (s.count("G") + s.count("C")) / len(s)
                     if not (gc_range[0] <= gc_content <= gc_range[1]):
                         continue
+
                     # 3. 3' 말단 안정성 및 GC Clamp
                     dg3 = sum(NN_PARAMS.get(s[-5:][j : j + 2], (0, 0))[0] for j in range(4))
                     if dg3 <= -10.0:
@@ -108,18 +109,63 @@ def generate_candidates(
                         continue
                     if s[-5:].count("G") + s[-5:].count("C") > 4:
                         continue
+
                     candidates.append(
                         {
                             "seq": s,
-                            "start": i,
-                            "end": i + k - 1,
+                            "start": i + 1,  # [수정] 1-based 시작점
+                            "end": i + k,    # [수정] 1-based 종료점 (inclusive)
                             "strand": strand,
                             "tm": tm,
                             "dg3": dg3,
                         }
                     )
         return candidates
 
+    # ==========================================
+    # 좌표 변환 및 매핑 유틸리티 추가
+    # ==========================================
+    def locate_template_in_genome(self, template_seq: str) -> Optional[Dict]:
+        """Stage 1: 입력된 템플릿 서열의 1-based 게놈 좌표 탐색"""
+        for ref in self.genome.references:
+            full_seq = self.genome.fetch(ref)
+            pos = full_seq.find(template_seq)
+            if pos != -1:
+                return {
+                    "chrom": ref,
+                    "genomic_start": pos + 1,  # 1-based 변환
+                    "strand": "+",
+                    "template_length": len(template_seq)
+                }
+            rev_seq = reverse_complement(template_seq)
+            pos = full_seq.find(rev_seq)
+            if pos != -1:
+                return {
+                    "chrom": ref,
+                    "genomic_start": pos + 1,  # 1-based 변환
+                    "strand": "-",
+                    "template_length": len(template_seq)
+                }
+        return None
+
+    def map_to_genomic_coords(self, primer: Dict, template_info: Dict) -> Dict:
+        """Stage 1.5: 1-based 로컬 좌표를 1-based 게놈 절대 좌표로 변환"""
+        p_start = primer["start"]
+        p_end = primer["end"]
+
+        if template_info["strand"] == "+":
+            primer["genomic_start"] = template_info["genomic_start"] + p_start - 1
+            primer["genomic_end"] = template_info["genomic_start"] + p_end - 1
+            primer["genomic_strand"] = primer["strand"]
+        else:
+            t_len = template_info["template_length"]
+            primer["genomic_start"] = template_info["genomic_start"] + (t_len - p_end)
+            primer["genomic_end"] = template_info["genomic_start"] + (t_len - p_start)
+            primer["genomic_strand"] = "-" if primer["strand"] == "+" else "+"
+
+        primer["chrom"] = template_info["chrom"]
+        return primer
+
     def local_db_filter(
         self,
         chrom: str,
@@ -129,31 +175,35 @@ def local_db_filter(
         intron_inclusion: bool = True,
         intron_size_range: Optional[Tuple[int, int]] = None,
     ) -> bool:
-        """Stage 2.2: 위치 및 구조 기반 필터링"""
+        """Stage 2.2: 위치 및 구조 기반 필터링 (게놈 절대 좌표 사용)"""
+        g_start = primer["genomic_start"]
+        g_end = primer["genomic_end"]
+        g_strand = primer["genomic_strand"]
 
-        # 1. SNP 필터링 (3' end strictness)
-        s_pos, e_pos = (
-            (primer["end"] - 4, primer["end"]) if primer["strand"] == "+" else (primer["start"], primer["start"] + 4)
-        )
+        # 1. SNP 필터링 (3' end strictness, inclusive search 적용)
+        s_pos, e_pos = (g_end - 4, g_end) if g_strand == "+" else (g_start, g_start + 4)
         self.cur.execute(
             "SELECT COUNT(*) FROM snp WHERE chrom=? AND pos BETWEEN ? AND ?",
             (chrom, s_pos, e_pos),
         )
         if self.cur.fetchone()[0] > 0:
             return False
-        # 2. 제한효소 필터링 (사용자 지정 리스트 반영)
+
+        # 2. 제한효소 필터링
         if restriction_enzymes:
             placeholders = ",".join(["?"] * len(restriction_enzymes))
             query = f"SELECT COUNT(*) FROM restriction_site WHERE chrom=? AND name IN ({placeholders}) AND NOT (end < ? OR start > ?)"
-            self.cur.execute(query, (chrom, *restriction_enzymes, primer["start"], primer["end"]))
+            self.cur.execute(query, (chrom, *restriction_enzymes, g_start, g_end))
             if self.cur.fetchone()[0] > 0:
                 return False
+
         # 3. Exon/Intron 구조 필터링
         self.cur.execute("SELECT start, end FROM exon WHERE chrom=? ORDER BY start", (chrom,))
         exons = self.cur.fetchall()
-        # Intron Inclusion 로직: 프라이머가 인트론 구간에 걸쳐 있는지 확인
+
+        # Intron Inclusion 로직
         is_in_intron = any(
-            primer["start"] > exons[i][1] and primer["end"] < exons[i + 1][0]
+            g_start > exons[i][1] and g_end < exons[i + 1][0]
             for i in range(len(exons) - 1)
         )
         if not intron_inclusion and is_in_intron:
@@ -162,84 +212,117 @@ def local_db_filter(
         # Intron Size 제한 확인
         if is_in_intron and intron_size_range:
             for i in range(len(exons) - 1):
-                if primer["start"] > exons[i][1] and primer["end"] < exons[i + 1][0]:
+                if g_start > exons[i][1] and g_end < exons[i + 1][0]:
                     i_size = exons[i + 1][0] - exons[i][1]
                     if not (intron_size_range[0] <= i_size <= intron_size_range[1]):
                         return False
+
         # Exon Junction Spanning 로직
         if junction_mode == "spanning":
             is_on_junction = any(
-                primer["start"] < e[1] and primer["end"] > exons[idx + 1][0]
+                g_start < e[1] and g_end > exons[idx + 1][0]
                 for idx, e in enumerate(exons[:-1])
             )
             if not is_on_junction:
                 return False
 
         return True
 
-    def specificity_check(
+    def filter_specific_primers(
         self,
-        chrom: str,
-        primer: Dict,
+        primers: List[Dict],
+        target_chrom: str,
         target_start: int,
         target_end: int,
         mispriming_library: bool = False,
         snp_exclusion: bool = False,
         splice_variant_handling: bool = False,
         max_hits=50,
         mismatch_cutoff=2,
-    ) -> bool:
-        """Stage 2.3: 게놈 전체 특이성 및 변이체 처리"""
+    ) -> List[Dict]:
+        """
+        Stage 2.3: 게놈 전체 특이성 일괄 검사 (I/O 병목 극적 최적화)
+        기존 specificity_check를 대체하며, 후보군 전체(List)를 받아 염색체 호출을 24회로 최소화합니다.
+        """
+        valid_primers = []
 
-        # 1. Mispriming Library (RepeatMasker 등 반복 서열 DB) 필터링
+        # 1. Mispriming Library (반복 서열) 필터링 - SQLite 쿼리 사전 수행
         if mispriming_library:
-            self.cur.execute(
-                "SELECT COUNT(*) FROM repeats WHERE chrom=? AND NOT (end < ? OR start > ?)",
-                (chrom, primer["start"], primer["end"]),
-            )
-            if self.cur.fetchone()[0] > 0:
-                return False
-        hits = 0
+            for p in primers:
+                self.cur.execute(
+                    "SELECT COUNT(*) FROM repeats WHERE chrom=? AND NOT (end < ? OR start > ?)",
+                    (p["chrom"], p["genomic_start"], p["genomic_end"]),
+                )
+                if self.cur.fetchone()[0] == 0:
+                    valid_primers.append(p)
+        else:
+            valid_primers = primers.copy()
+
+        # 2. 상태 추적용 딕셔너리 구성
+        primer_pool = {p["seq"]: p for p in valid_primers}
+        hit_counts = {p["seq"]: 0 for p in valid_primers}
+
+        # 3. 루프 역전: 염색체를 한 번만 불러오고, 메모리 상에서 남은 모든 프라이머 스캔
         for ref in self.genome.references:
+            if not primer_pool:
+                break  # 모든 프라이머가 탈락했다면 스캔 즉시 종료
+
             full_seq = self.genome.fetch(ref)
-            for search_seq in [primer["seq"], reverse_complement(primer["seq"])]:
-                pos = full_seq.find(search_seq)
-                while pos != -1:
-                    # 의도된 타겟 구간 제외
-                    if ref == chrom and target_start <= pos <= target_end:
-                        pos = full_seq.find(search_seq, pos + 1)
-                        continue
 
-                    # 2. Splice Variant Handling
-                    if splice_variant_handling:
-                        self.cur.execute(
-                            "SELECT transcript_id FROM exon WHERE chrom=? AND start <= ? AND end >= ?",
-                            (ref, pos, pos + len(search_seq)),
-                        )
-                        if self.cur.fetchone():
-                            # 동일 유전자의 다른 전사체라면 off-target에서 제외
-                            pos = full_seq.find(search_seq, pos + 1)
-                            continue
-                    off_target = full_seq[pos : pos + len(primer["seq"])]
-
-                    # 3. SNP Exclusion
-                    if snp_exclusion:
-                        self.cur.execute(
-                            "SELECT COUNT(*) FROM snp WHERE chrom=? AND pos BETWEEN ? AND ?",
-                            (ref, pos, pos + len(search_seq)),
-                        )
-                        if self.cur.fetchone()[0] > 0:
+            for p_seq in list(primer_pool.keys()):
+                for search_seq in [p_seq, reverse_complement(p_seq)]:
+                    pos = full_seq.find(search_seq)
+
+                    while pos != -1:
+                        pos_1based = pos + 1
+                        end_1based = pos + len(search_seq)
+
+                        # 의도된 타겟 구간 무시
+                        if ref == target_chrom and target_start <= pos_1based <= target_end:
                             pos = full_seq.find(search_seq, pos + 1)
                             continue
-                    # 3' 말단 미스매치 정밀 검사
-                    mm = needleman_wunsch_mismatch(primer["seq"][-10:], off_target[-10:])
-                    if mm < mismatch_cutoff:
-                        return False
-                    hits += 1
-                    if hits > max_hits:
-                        return False
-                    pos = full_seq.find(search_seq, pos + 1)
-        return True
+
+                        # DB 변이체 필터링 (일치 타겟 발견 시에만)
+                        if splice_variant_handling:
+                            self.cur.execute(
+                                "SELECT transcript_id FROM exon WHERE chrom=? AND start <= ? AND end >= ?",
+                                (ref, pos_1based, end_1based),
+                            )
+                            if self.cur.fetchone():
+                                pos = full_seq.find(search_seq, pos + 1)
+                                continue
+
+                        if snp_exclusion:
+                            self.cur.execute(
+                                "SELECT COUNT(*) FROM snp WHERE chrom=? AND pos BETWEEN ? AND ?",
+                                (ref, pos_1based, end_1based),
+                            )
+                            if self.cur.fetchone()[0] > 0:
+                                pos = full_seq.find(search_seq, pos + 1)
+                                continue
+
+                        # 3' 말단 미스매치 정밀 검사
+                        off_target = full_seq[pos : pos + len(p_seq)]
+                        mm = needleman_wunsch_mismatch(p_seq[-10:], off_target[-10:])
+
+                        if mm < mismatch_cutoff:
+                            # 치명적 Off-target 발견 시 즉시 탈락
+                            del primer_pool[p_seq]
+                            break 
+
+                        hit_counts[p_seq] += 1
+                        if hit_counts[p_seq] > max_hits:
+                            del primer_pool[p_seq]
+                            break
+
+                        pos = full_seq.find(search_seq, pos + 1)
+
+                    # 현재 프라이머가 탈락했다면, Reverse Complement 검사 등 내부 루프 완전히 중단
+                    if p_seq not in primer_pool:
+                        break
+
+        # 안전성이 검증된 프라이머들만 반환
+        return list(primer_pool.values())
 
     def pair_primers(
         self,
@@ -254,7 +337,8 @@ def pair_primers(
         pairs = []
         for f in fwd:
             for r in rev:
-                size = r["start"] - f["end"]
+                # [수정] 생물학적으로 올바른 Product Size 계산 (1-based 기준)
+                size = r["end"] - f["start"] + 1
                 if not (product_range[0] <= size <= product_range[1]):
                     continue
 
@@ -277,4 +361,4 @@ def pair_primers(
                         "penalty": penalty,
                     }
                 )
-        return sorted(pairs, key=lambda x: x["penalty"])
+        return sorted(pairs, key=lambda x: x["penalty"])

docs/prompts/11주차/algorithm_optimization.md

@@ -0,0 +1,31 @@
+# 프라이머 디자인 알고리즘 최적화 및 좌표 보정
+
+## 1. 배경 및 목적
+
+- 사용자가 입력한 템플릿 서열(Local 좌표)과 게놈 DB(Genomic 절대 좌표) 간의 좌표 불일치 문제 해결
+- 1-based 기준인 생물학적 데이터를 0-based 파이썬 인덱스로 처리하며 발생하는 오차 수정
+- 전수 게놈 스캔 시 발생하는 심각한 I/O 병목 현상을 해결하여 발표 시연이 가능한 수준으로 성능 개선
+
+## 2. 프라이머 (User Input)
+
+```text
+(좌표계 문제 제기)
+이 코드를 보면 primer design 알고리즘을 설계한 것을 알 수 있어. 문제는 여기서 UI에서 입력값으로 fasta 파일이나 txt 파일을 넣을거야. 이때, template sequence가 해당 organism species의 genome을 읽어들여서 template sequence의 genome 상에서의 좌표를 파악해서 아래 코파일럿이 제기한 문제를 해결할려고 해. 기존 BLAST designer 등에서 사용하는 방법들을 사용해서 로직 코드를 보완해줘.
+-제기된 문제: PrimerDesigner.py는 게놈 좌표/DB 좌표 기반으로 동작하는 전제인데, 현재 입력(템플릿 서열/검색 범위)이 템플릿 좌표로 들어오면 필터가 어긋납니다.
+
+(성능 문제 제기)
+근데 genome 상 좌표를 그렇게 찾으면 시간이 너무 오래 걸릴 거 같은데? 로직 자체를 효율적으로 수정해줬으면 해. API한테 전달 받는 것도, 전달 하는 것도 변함은 없어야 해.
+
+```
+
+## 3. AI 응답 요약 (AI Output)
+
+* **좌표계 동기화**: `locate_template_in_genome` 메서드를 추가하여 템플릿의 게놈 내 위치를 찾고, 모든 로컬 좌표를 1-based 게놈 절대 좌표로 매핑하는 로직 구현.
+* **I/O 최적화 (Batch Scan)**: 기존에 프라이머마다 염색체 파일을 반복 호출하던 방식에서, 염색체를 한 번 로드할 때 모든 프라이머 후보를 일괄 검사하는 루프 역전(Loop Inversion) 기법 적용.
+* **정밀도 향상**: SNP, Exon 필터링 시 3' 말단(Inclusive range) 처리를 1-based 기준으로 정확히 계산하도록 수정.
+
+## 4. 결과 및 적용 (Result)
+
+* **적용 사항**: `app/algorithms/PrimerDesigner.py` 내 `generate_candidates`, `local_db_filter`, `pair_primers` 등 핵심 로직 전면 수정.
+* **성능 변화**: 수백 개의 후보군에 대한 특이성 검사 시간이 디스크 I/O 최적화를 통해 획기적으로 단축됨.
+* **영향**: API 엔드포인트의 구조 변경 없이 알고리즘 내부 수정만으로 요구사항을 충족하여 프론트엔드/API 파트와의 정합성 유지.